Farmer.cpp-Dateireferenz

#include <vector>
#include <sstream>
#include <iostream>
#include <fstream>
#include <string>
#include <mutex>
#include <chrono>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/transport/TSocket.h>
#include <thrift/transport/TTransportUtils.h>
#include "gen-cpp/Solver.h"
#include "include/Buffer.hpp"
#include "include/Connection.hpp"

Makrodefinitionen
#define	CONFIGFILE   "farmer.conf"
#define	INITIALOPT   0

Funktionen
void	fillInitialFix (void)
bool	isValid (vector< bool > fix)
void	branch (vector< bool > fix)
void	workerFunction (SolverClient *remote)
void	farmerFunction (void)
void	distributor (string fileName)
int	main (int argc, char *argv[])

Variablen
Instance	in
	diese Daten werden später an die Worker gesendet Mehr ...
int	workerCnt
int	vorbelegung
mutex	muSol
SolveResult	sol
mutex	muJobs
long long int	jobs
long long int	jobAdd
Buffer< vector< bool > >	jobBuffer
bool	killWorker
chrono::steady_clock::time_point	go
	Im C++2011 Standard gibt es nicht nur mutex, sondern auch bessere Zeitmessungs-Funktionen. Mehr ...
chrono::steady_clock::time_point	fin
vector< vector< bool > >	initalFix
	In dieser Variablen werden alle Vorbelegungen inital abgespeichert. Mehr ...

Makro-Dokumentation

#define CONFIGFILE   "farmer.conf"

#define INITIALOPT   0

Dokumentation der Funktionen

void branch

(

vector< bool >

fix

)

Diese Funktion fügt an eine Vorbelgung eine weitere Vorbelegung hinzu. Wenn diese neue Vorbelegung gültig ist, wird sie dem JobBuffer begefügt. Diese Funktion ist ein Produzent, da sie in den Buffer Jobs ablegt. Es kann etwas verwirrend sein, weil ein Worker-Thread in diesem Beispiel im Grunde sowohl Konsument als auch Produzent ist, weil in einem Worker die branch() Funktion aufgerufen wird. Diese Vermischung ist unüblich.

Parameter

fix	ist ein Bool-Vektor mit einer Projekt-Vorbelegung

Rückgabe

nichts

{
    int u = fix.size();
    fix.resize(u + vorbelegung);
    
    // jede neue Vorbelegung an die Bisherige anhängen
    for (vector<bool> & val : initalFix) {
        for (int i = 0; i < vorbelegung; ++i) fix[u + i] = val[i];
        // nur gültige Vorbelegungen (Budget-Grenzen!) werden Buffer beigefügt
        if (isValid(fix) == true) { 
            muJobs.lock();
            ++jobs;
            jobBuffer.push(fix);
            muJobs.unlock();
        }
    }
    muSol.lock();
    fin = chrono::steady_clock::now();
    cout << chrono::duration_cast<chrono::seconds>(fin-go).count();
    cout << "s, branch()" << endl;
    muSol.unlock();
}

void distributor

(

string

fileName

)

Diese Funktion stellt die Verbindung zu allen Workern her und startet zur asynchronen Verarbeitung der Verbindungen Worker-Threads (= Proxies). Ein Farmer-Thread, der primär eine überwachende Funktion über die Worker-Threads und den jobBuffer hat, wird ebenfalls gestartet.

Parameter

fileName

Der Name der Konfigurations-Datei, in der alle Infos zur Verteilung (engl.: Distribution) stehen

Rückgabe

nichts

{
    // config einlesen
    ifstream file(fileName);
    if (file.is_open() == false) {
        throw "missing file or other file error.";
    }
    string temp, rawdata;
    while (getline(file, temp)) rawdata += temp;
    file.close();
    stringstream f(rawdata);

    f >> workerCnt;
    f >> vorbelegung;
    f >> in._timeoutMs;
    
    // einmal zu Anfang alle möglichen Vorbelegungen erzeugen
    fillInitialFix();
    
    vector<Connection> conn(workerCnt);
    for (int i=0; i < workerCnt; ++i) {
        char ip[39]; //ipv6 ?
        int port;
        f >> ip;
        f >> port;
        cout << "[" << ip << "]:" << port << endl;
        conn[i].socket = PtrTran(new TSocket(ip, port));
        conn[i].transport = PtrTran(new TBufferedTransport(conn[i].socket));
        conn[i].protocol = PtrProt(new TBinaryProtocol(conn[i].transport));
        conn[i].remote = new SolverClient(conn[i].protocol);
        // Verbindung zum Worker oeffnen
        conn[i].transport->open();
        // Instanz uebermitteln
        conn[i].remote->setInstance(in);
        // auf Daten im jobBuffer wartenden Worker-Proxy-Thread erzeugen
        conn[i].proxy = new thread(workerFunction, conn[i].remote);
    }
    
    // den Farmer-Thread starten
    thread tFarmer = thread(farmerFunction);
    // auf das Edne des Farmer-Thread warten
    tFarmer.join();
    // auf das Ende alle Worker-Proxy-Threads warten
    for (int i=0; i < workerCnt; ++i) {
        conn[i].proxy->join();
    }
    // Verbindung schlie/3en und Speicher fuer (nicht shared) Pointer freigeben
    for (int i=0; i < workerCnt; ++i) {
        conn[i].transport->close();
        delete conn[i].remote;
        delete conn[i].proxy;
    }
}

void farmerFunction

(

void

)

Diese Funktion läuft auch als Thread. Sie ist ein Produzent weil sie mit push() in den Buffer Jobs ablegt. Eine kleine Endlos-while-Schleife prüft alle 500000 Microsekunden, ob bereits alle Jobs aus den Buffer erledigt wurden. Ohne das usleep würde dieser Farmer-Thread viel CPU-Zeit kosten.

Rückgabe

nichts

{
    vector<bool> emptyFix;
    // inital erste Vorbelegungen/Jobs erzeugen
    branch(emptyFix);
    
    while( true ) {
        muJobs.lock();
        if(jobs == 0) {
            muJobs.unlock();
            break;
        }
        muJobs.unlock();
        usleep(500000);
    }
    killWorker = true;
    /* Mit leeren Jobs die Worker aus dem "auf den buffer warten" aufwecken,
     * damit diese sich beenden können */
    for (int i = 0; i < workerCnt; ++i) {
        jobBuffer.push(emptyFix);
    }
}

void fillInitialFix

(

void

)

Um in der Funktion branch() nicht jedes mal einen binären Vektor hoch zu zählen, wird dies einmal zu Anfang mit dieser Funktion gemacht und die Werte dem bestehenden fix-Vector nur noch beigefügt. Statt Parameter arbeitet die funktion mit der globalen Variable vorbelegung.

Rückgabe

nichts; arbeitet mit globalen Variablen jobAdd und initialFix

{
    jobAdd = (long long int) 1 << vorbelegung;
    vector<bool> counter(vorbelegung, true);
    initalFix.resize(jobAdd);
    for (long long int n = 0; n < jobAdd; ++n) {
        initalFix[n] = counter;
        long long int tmp = 1;
        for (int i = 0; i < vorbelegung; ++i) {
            if (((n+1) % tmp) == 0)
                counter[i] = !(counter[i]);
            tmp *= 2;
        }
    }
}

bool isValid

(

vector< bool >

fix

)

Die Funktion isValid() testet einen binären Vektor, ob dieser eine gültige Projekt-Vorbelegung enthält.

Parameter

fix	ist ein Bool-Vektor mit einer Projekt-Vorbelegung

Rückgabe

bool-Wert ist true, wenn eine Vorbelegung die Budgetgrenzen einhält.

{
    int fi = fix.size();
    int T  = in._budgets.size();

    int price;
    for (int t = 0; t < T; ++t) {
        price = 0;
        for (int i = 0; i < fi; ++i) {
            if (fix[i]) price += in._costs[t][i];
            
            if (price > in._budgets[t]) return false;
        }
    }
    return true;
}

int main	(	int	argc,
		char *	argv[]
	)

Der "Farmer" ist die Client Anwendung, die die in kleine Teilprobleme aufgeteilten Aufgaben für glpk auf die Worker (Server) verteilt.

{
    killWorker = false;
    sol._optimum = INITIALOPT;
    
    if (argc < 2) {
        cout << "usage: " << argv[0] << " <instance file>" << endl;
        return 1;
    }
    
    // Instanz aus Datei einlesen ------------
    ifstream file(argv[1]);
    if (file.is_open() == false) {
        throw "missing file or other file error.";
    }
    string temp, rawdata;
    while (getline(file, temp)) rawdata += temp;
    file.close();
    stringstream f(rawdata);

    int N, T, opt;
    
    f >> N; f >> T; f >> opt;
    in._profits.resize(N);
    for (int n = 0; n < N; ++n) {
        f >> in._profits[n];
    }
    in._costs.resize(T);
    for (int t = 0; t < T; ++t) {
        in._costs[t].resize(N);
        for (int n = 0; n < N; ++n) {
            f >> in._costs[t][n];
        }
    }
    in._budgets.resize(T);
    for (int t = 0; t < T; ++t) {
        f >> in._budgets[t];
    }
    
    // verteilen und loesen ---------
    go = chrono::steady_clock::now();
    distributor(CONFIGFILE);
    fin = chrono::steady_clock::now();
    cout << chrono::duration_cast<chrono::seconds>(fin-go).count() << "s ";

    // ausgabe ----------------------
    for (bool val : sol._projects) {
        val ? cout << "I" : cout << "o";
    }
    cout << " Optimum: " << sol._optimum << endl;
    return 0;
}

void workerFunction

(

SolverClient *

remote

)

Diese Funktion wird als Thread benutzt und ist ein Proxy. Zu jeder Verbindung zu einem Service eines Workers existiert auf dem Farmer so ein Thread. Solange wie dieser aus jobBuffer Jobs holen kann und die Variable killWorker keinen Abbruch erzeugt, werden Jobs an den Service der Worker zum lösen - also solve() - gesendet. Dieser Thread ist ein Konsoment und durch eine Semaphore in Buffer geschieht beim pop() aufruf nichts, solange im Buffer nichts ist.

Parameter

remote

Repräsentiert den Service des Workers

Rückgabe

nichts

{
    while (true) {
        vector<bool> fix = jobBuffer.pop();
        if (killWorker == true) return;
        
        SolveResult _return;
        remote->solve(_return, fix, sol._optimum);
        
        if (_return._flag == Flag::OK) {
            muSol.lock();
            if (sol._optimum < _return._optimum) {
                sol = _return;
                fin = chrono::steady_clock::now();
                cout << chrono::duration_cast<chrono::seconds>(fin-go).count();
                cout << "s, new Best: " ;
                cout << _return._optimum << endl; 
            }
            muSol.unlock();
        }
        if (_return._flag == Flag::TIMEOUT) {
            branch(fix);
        }
        muJobs.lock();
        --jobs;
        muJobs.unlock();
    }
}

Variablen-Dokumentation

chrono::steady_clock::time_point fin

chrono::steady_clock::time_point go

Im C++2011 Standard gibt es nicht nur mutex, sondern auch bessere Zeitmessungs-Funktionen.

Instance in

diese Daten werden später an die Worker gesendet

vector<vector<bool> > initalFix

In dieser Variablen werden alle Vorbelegungen inital abgespeichert.

long long int jobAdd

Buffer<vector<bool> > jobBuffer

Dieser Buffer nimmt alle Jobs auf - also alle Projekt-Vorbelegungen, bei denen die Worker mit glpk eine optimale Lösung finden sollen.

long long int jobs

bool killWorker

mutex muJobs

Die Variable jobs wird ebenfalls von allen Threads ausgelesen und verändert. Dieser kritische Bereich muss ebenfalls geschützt werden.

mutex muSol

Da alle Threads (= unsere Worker-Proxies) auf die sol Variable zugreifen können, muss dieser Zugriff mit einer Mutex-Semaphore geschützt werden.

SolveResult sol

int vorbelegung

int workerCnt