Multithreaded Server - Design Guidelines?

Volkard schrieb:

Zeig mal den Code, dann teste ich hier auch mal, falls er auf Linux läuft.

Da ich hier fuer den ThreadPool interne libs nutze, kann ich leider nicht den ganzen Code posten. Aber zumindest die Stueckwerke gehen (bitte nicht auf Schoenheit achten, ist schnell hingehackt)

Hauptprogramm

#include <ctime>
#include <algorithm>

#include <vector>
#include <iostream>
#include <thread>
#include <future>

#include "ThreadPool.h"
#include "Semaphore.h"
#include "PerformanceCounterWin32.h"

#include "DummyFunctorThreadPool.h"
#include "DummyFunctorAsync.h"

using namespace std;

const size_t numClients = 3000;
const size_t numActionCycles = 100;
const size_t numBurnCycles = 10000;
const size_t vecSize = 100;

const int startDelayInMilliseconds = 2000;
const int idleTimeInMilliseconds = 20;

Semaphore semTP(0);
Semaphore semAsync(0);

void startAsync()
{
	vector<shared_ptr<thread>> threads; // these threads will execute the DummyFunctorAsync operator() method

	for(int i = 0; i<numClients; ++i)
	{
		threads.emplace_back(make_shared<thread>(DummyFunctorAsync(&semAsync)));
	}
	for(int i = 0; i<numClients; ++i)
	{
		threads[i]->join();
	}

	cout << "started" << endl;
	semAsync.acquire(numClients);
	cout << "finished" << endl;
}

void startThreadPooled()
{
	ThreadPool tp(thread::hardware_concurrency()*2);

	// in a ThreadPool approach, a worker will be terminated when a task is finished and need to be
	// enqueued again
	for(int i = 0; i<numClients*numActionCycles; ++i)
	{
		tp.process(move(unique_ptr<DummyFunctor>(new DummyFunctor(&semTP))));
	}

	cout << "started" << endl;
	semTP.acquire(numClients*numActionCycles);

	cout << "finished" << endl;
}

int main() 
{
	PerformanceCounterWin32 perfCount;

	perfCount.startMeasurement();
	{
		startThreadPooled();
	}

	perfCount.queryCounter();
	{
		startAsync();
	}

	perfCount.queryCounter();

	cout << "ThreadPool took " << perfCount.getDeltaInSeconds(1, 2) << "\n";
	cout << "ASync took " << perfCount.getDeltaInSeconds(0, 1) << "\n";
	cout << endl;
}

DummFunctor fuer den ThreadPool
.h

#pragma once

#include <vector>
#include "ThreadPool.h"
#include "Semaphore.h"

#include "constants.h"

class DummyFunctor : public WorkItem
{
public:
	// intializes semaphore plus vector
	DummyFunctor(Semaphore* sem);
	DummyFunctor(DummyFunctor&& rhs);
	DummyFunctor& operator=(const DummyFunctor& rhs);

	// execute some actions in order to simulate a client
	void operator()();
protected:
	// method that does the processing using both integer and floating point
	void burnSomeCPUCycles();

	double sum;
	std::vector<double> vec;
	Semaphore* sem;
};

.cpp dazu

#include "DummyFunctorThreadPool.h"

#include <algorithm>
#include <random>
#include <vector>
#include <iostream>

using namespace std;

DummyFunctor::DummyFunctor(Semaphore* sem) : sum(0.0), vec(vecSize), sem(sem)
{
	mt19937 rng;

	generate(vec.begin(), vec.end(), rng);
}

DummyFunctor::DummyFunctor(DummyFunctor&& rhs)
{
	this->sum = rhs.sum;
	this->vec = rhs.vec;
	this->sem = rhs.sem;
}
DummyFunctor& DummyFunctor::operator=(const DummyFunctor& rhs)
{
	this->sum = rhs.sum;
	this->vec = rhs.vec;
	this->sem = rhs.sem;
	return *this;
}

void DummyFunctor::operator()()
{
	burnSomeCPUCycles();
	sem->release();
}

void DummyFunctor::burnSomeCPUCycles()
{
	//		this_thread::sleep_for(chrono::milliseconds(idleTimeInMilliseconds));
	for (int i=0; i<numBurnCycles; ++i)
	{
		sum += log(sin(vec[i%vecSize]));
	}
}

Dasselbe fuer den ASync Teil
.h

#pragma once

#include <vector>
#include "ThreadPool.h"
#include "Semaphore.h"

#include "constants.h"

#include "DummyFunctorThreadPool.h"

class DummyFunctorAsync : public DummyFunctor
{
public:
	DummyFunctorAsync(Semaphore* sem);
	DummyFunctorAsync(DummyFunctorAsync&& rhs);
	DummyFunctorAsync& operator=(const DummyFunctorAsync& rhs);

	void operator()();
};

.cpp

#include "DummyFunctorAsync.h"

using namespace std;
using namespace neotaix;

DummyFunctorAsync::DummyFunctorAsync(Semaphore* sem) : DummyFunctor(sem)
{
}

DummyFunctorAsync::DummyFunctorAsync(DummyFunctorAsync&& rhs) : DummyFunctor(rhs)
{
}

void DummyFunctorAsync::operator()()
{
	// wait for all threads to start up. This is required here because otherwise
	// processing will start immediately leading to a contention where not all threads 
	// can started completely before the first are finishing
	this_thread::sleep_for(chrono::milliseconds(startDelayInMilliseconds));
	// the thread stays active until all actions are done
	for(int i = 0; i < numActionCycles; ++i)
	{
		burnSomeCPUCycles();
		this_thread::sleep_for(chrono::milliseconds(idleTimeInMilliseconds));
	}
	sem->release();
}

und der header zum Bekanntmachen der Konstanten

#pragma once
extern const size_t numClients;
extern const size_t numActionCycles;
extern const size_t numBurnCycles;
extern const size_t vecSize;
extern const int idleTimeInMilliseconds;
extern const int startDelayInMilliseconds;

Und noch die Deklaration fuer den WorkItem

/**
     * A WorkItem object is an object that provides void operator(void)
     */ 
    class WorkItem : public unary_function<void, void>
    {
    public:
		/**
		 * Overload this method in order to implement your own concurrent processing. Use class
		 * member variables to transport parameters into the method.
		 * If exceptions might be thrown from this method, you should use futures to handle that. Otherwise,
		 * the threadpool will catch and rethrow it.
		 */
        virtual result_type operator()(argument_type) = 0;
    };

hustbaer

DOSen schrieb:

Ergebnis: Wie erwartet gibt es bei der Verarbeitungsleistung praktisch keinen Unterschied. In beiden Faellen war die CPU durch den Prozess mit 94%-97% bzw 96%-99% ausgelastet. Aber: Mit 3000 Threads ist die Maschine praktisch nicht mehr benutzbar gewesen. Habe testweise im Hintergrund einen InetRadioStream laufen lassen. Bei 3000 Threads hat man nur noch abgehackte Fetzen hoeren koennen, beim Threadpool lief alles ohne dropouts weiter.

Und wie sah der Throughput der beiden Varianten aus?
DAS wäre nämlich viel interessanter.

DOSen schrieb:

Fazit: Fuer so etwas am besten also einen Master-Worker Ansatz nehmen.

Dann dreh die Thread-Priorität der Server-Threads auf -1, und die Sache ist gegessen.

hustbaer schrieb:

Und wie sah der Throughput der beiden Varianten aus?

Annähernd gleich. Allerdings war die Latenz beim Starten eines neuen Threads bzw. einqueuen eine Clients bei der Variante mit einem Thread pro Client viel höher als bei der ThreadPool Variante.

hustbaer schrieb:

Dann dreh die Thread-Priorität der Server-Threads auf -1, und die Sache ist gegessen.

Habe ich mittels SetThreadPriority(threads.back()->native_handle(), THREAD_PRIORITY_BELOW_NORMAL); ausprobiert. Das ist zwar deutlich besser, aber vom Ansprechverhalten immer noch schlechter als die ThreadPool Variante.

Aber Danke für den Hinweis. Da hatte ich nicht dran gedacht. Im Gegenzug gibt es natürlich die Option auch beim ThreadPool, plus man kann die Poolgröße ja auch noch hochdrehen.

Oder hast Du vielleicht noch eine Alternative in petto?

hustbaer

Oder hast Du vielleicht noch eine Alternative in petto?

Alternative für was?

Für Fälle wo der Server für die Clients nur lauter Dinge zu machen hat die keine hohe Rechenleistung im Connection-Thread erfordern kann man denke ich nix mehr drehen. Wenn da der "1 Thread pro Connection" Server in die Knie geht, dann ist man mit der Variante wohl am Limit.

Bei Fällen wo der Server in den Connection Threads richtig viel rechnen würde, könnte ich mir aber vorstellen dass man die beiden Varianten hübsch kombinieren kann.
Also die ganzen "billigen" Sachen macht man direkt im Connection Thread, und aufwendigere Berechnungen lagert man in einen Thread Pool aus.
Also quasi

Reply Connection::HandleFooRequest(Foo foo)
{
   LogSomething(foo);
   if (!VerifySomething(foo, m_someOtherValue))
   {
       LogSomethingElse(foo);
       return MakeErrorReply("blub");
   }

   // ...

   DoSomeStuffThatMightBlockTheThreadForSeveralSeconds();

   // ...

   auto fooData = m_pool.Queue([&](){ return LongRunningCalculation(foo); });

   // ...
   // ...

   auto fooData2 = m_pool.Queue([&](){ return AnotherHeavyCalculation(foo); });

   // ...
   // ...

   return MakeFooReply(fooData.get(), fooData2.get());
}

Das hat den Vorteil dass man viel Code der nur sehr lästig asynchron zu machen wäre im Connection Thread lassen kann, aber trotzdem nicht tausende Threads hat die permanent die CPU mit rechenintensiven Sachen hämmern.

Ich schätze mal das sollte sich sehr positiv auf die Responsiveness des Servers auswirken.

ps: Ich hab aber den Thread hier nicht sehr aufmerksam gelesen. Also falls meine Antwort total unpassend ist bitte einfach ignorieren (oder auch gerne kurz darauf hinweisen wieso).

volkard

DOSen schrieb:

Volkard schrieb:

Zeig mal den Code, dann teste ich hier auch mal, falls er auf Linux läuft.

Da ich hier fuer den ThreadPool interne libs nutze…

Hab den Code mit 3000 Threads mal getestet.
Linux 64Bit i7 16G.
Schlechte Ansprechbarkeit bemerke ich eigentlich nicht, aber ich kriege nur 32273 Threads auf.

hustbaer

DOSen schrieb:

hustbaer schrieb:

Und wie sah der Throughput der beiden Varianten aus?

Annähernd gleich. ...

ps: Danke für die Info

Kellerautomat

Ich weiss immer noch nicht, wie ich die Simulation von IO unabhaengig bekommen soll. Ich benutze fuer IO boost.asio, d.h. das Threading wird hier mittels asio's io_service erledigt. Wenn ich allerdings einen Thread pro Welt haben moechte, funktioniert das nicht, weil die Welt dann ebenfalls ueber den io_service bescheid wissen muss.

Ich habe einen Scheduler, der so aussieht:

struct WorldScheduler
{
	typedef std::function<void(World&)> Task;

	WorldScheduler(World& world)
		: world_(&world)
		, run_(true)
		, thread_(&WorldScheduler::executeTasks, this)
	{}

	~WorldScheduler()
	{
		run_ = false;
		thread_.join();
	}

	void spawnTask(Task task)
	{
		scheduledTasks_.push(std::move(task));
	}

	void spawnRepeatedTask(Task task)
	{
		repeatedTasks_.push_back(std::move(task));
	}

private:
	void executeTasks()
	{
		Stopwatch sw;

		while(run_)
		{
			tick();

			for(Task task; sw.expired() < TICK_DURATION && scheduledTasks_.try_pop(task);)
				task(*world_);

			auto expired = sw.reset();

			if(expired < TICK_DURATION)
				std::this_thread::sleep_for(TICK_DURATION - expired);
		}
	}

	void tick()
	{
		for(auto& task : repeatedTasks_)
			task(*world_);
	}

	World* world_;
	atomic<bool> run_;
	concurrent_vector<Task> repeatedTasks_;
	concurrent_queue<Task> scheduledtasks_;

	std::thread thread_;
};

Die Welt braucht hier volle Kontrolle ueber das Scheduling, u.A. aufgrund der Server-Ticks, die 20 mal pro Sekunde ausgefuehrt werden sollen. Kann man das irgendwie sinnvoll mit boost.asio zusammenbringen?

Volkard schrieb:

Schlechte Ansprechbarkeit bemerke ich eigentlich nicht, aber ich kriege nur 32273 Threads auf.

Interessant. Dann scheint das ja eher am Windows Scheduler zu liegen. Hast Du mal, wenn die 32k Threads laufen einen Audiostream laufen lassen? Das ist eigentlich ein guter Test, ob das System noch damit zurecht kommt.

Unter Windows habe ich extra 1s Wartezeit eingebaut, bevor die Threads zu rechnen anfangen, ansonsten wurde das Aufstarten weiterer Threads so langsam, dass ich nie über ein paar Hundert glecihzeitig laufender Threads gekommen bin.
Ich hatte zuerst vermutet, dass die Threads schneller fertig werden als neue hinzu kommen, aber das bleibt gleich, auch wenn man die Werte entsprechend ändert.

hustbaer schrieb:

Also die ganzen "billigen" Sachen macht man direkt im Connection Thread, und aufwendigere Berechnungen lagert man in einen Thread Pool aus.

Ja, das ist wahrscheinlich ein brauchbarer Ansatz. Wenn ich allerdings Volkards Messungen in Betracht ziehe, scheint es ja vor allem am OS zu liegen. Sprich, bei vielen Clients ist der Overhead für das Thread Scheduling unter Windows (in Zusammenhang mit dem Design des Codes) problematisch.

Ich habe hier leider kein Linux System im Zugriff auf dem ich das ausprobieren könnte...

volkard

DOSen schrieb:

Volkard schrieb:

Schlechte Ansprechbarkeit bemerke ich eigentlich nicht, aber ich kriege nur 32273 Threads auf.

Interessant. Dann scheint das ja eher am Windows Scheduler zu liegen. Hast Du mal, wenn die 32k Threads laufen einen Audiostream laufen lassen? Das ist eigentlich ein guter Test, ob das System noch damit zurecht kommt.

Ich habe normal weitergesurft, im Forum geschaut, und ein youtube-Video lief dazu.

volkard schrieb:

Ich habe normal weitergesurft, im Forum geschaut, und ein youtube-Video lief dazu.

Beeindruckend. Entweder ich mache noch etwas grundlegend falsch oder der Windows-Scheduler ist wirklich schlecht.

Jodocus

DOSen schrieb:

volkard schrieb:

Ich habe normal weitergesurft, im Forum geschaut, und ein youtube-Video lief dazu.

Beeindruckend. Entweder ich mache noch etwas grundlegend falsch oder der Windows-Scheduler ist wirklich schlecht.

Wo sind denn X-tausend Threads noch sinnvoll? Egal bei welchem OS, so ziemlich jedes wird dann einen Großteil der Zeit beim Kontextwechsel verbraten.

hustbaer

@Jodocus

Jodocus schrieb:

so ziemlich jedes wird dann einen Großteil der Zeit beim Kontextwechsel verbraten.

Kontextwechsel sind aber gar nicht das Problem.
Das "Problem" hierbei ist dass der Scheduler jedem aktiven Thread gleich viel Zeit zuweist - statt jedem Prozess (bzw. Prozessgruppe) mit aktiven Threads gleich viel Zeit zuzuweisen.

Es gibt allerdings auch Scheduler die letzteres machen - mit denen kann man dann ganz normal weiterarbeiten während der Server mit den 10K+ Threads läuft.

Kellerautomat

Ich hab jetzt einen Ansatz gefunden, allerdings ist er suboptimal. Ich koennte pro Welt einen io_service verwenden und in den WorldScheduler ein .poll() als repeated Task reinwerfen. Problem dabei: mit asio's socket funktioniert das nicht, weil man da den io_service nicht neu zuweisen kann, daher muss ich einen posix::stream_descriptor verwenden und den accept() Code mit dem BSD-API selbst schreiben.

Hat hier jemand ne bessere Idee?

TyRoXx

Sind die paar tausend Nachrichten pro Sekunde wirklich relevant? Parallelisierung würde sich erst dann lohnen, wenn du die Kapazität von Asio überschreiten würdest.

Finde heraus, was wirklich langsam ist. Das ist vermutlich die Kernlogik des Spiels mit ihren vielen Suchen in den Datenstrukturen für die Spielwelt. Ich kann mir vorstellen, dass das Original da relativ naiv arbeitet und auch wegen Java nicht auf Cache optimiert ist. Diese Suchen kann man möglicherweise auch parallel durchführen, indem man die Welt auf Prozessorkerne aufteilt.

Kellerautomat

Es geht nicht darum, IO zu parallelisieren, sondern die Simulation. Dass die Clients dem jeweiligen Thread ihrer Welt zugeteilt werden, ist nur eine Folge davon.

hustbaer

@Kellerautomat
Hast du dir mal die ASIO Strands angesehen?

Kellerautomat

strands bringen mir nichts.

Edit: Vielleicht kann man damit doch was machen, aber schoen ist anders.

volkard

Kellerautomat schrieb:

Ich hab kein Problem mit Automaten. Ich muss so oder so Event-Handling irgendwie bauen, da duerfte das relativ egal sein.

Ich fahre sowas von ab auf kleine Funktionen und Destruktoren und RAII und Exceptions…
Erzeuge ich einen LoginDialog, dann wird er beim Kunden auch sichtbar. Wenn ich den LoginDialog warumauchimmer delete, dann verschwindet der auch. Erkenne ich einen cheat, dann werfe ich eine Exception. Ganz normales Stack-Unwinding kommt. Der hochgehobene Gegenstand im Rollenspiel verliert seinen Griff/Lock, mit dem ich ihn anpackte (und davor schütze, daß ein Anderer ihn hochebt), und plumpst runter. Ich muss nicht mehr wie beim Automaten jede Zustandsändernde Aktion wie Objektaufheben/Zauberbuchöffnen/Handelsdialogöffnen/Holzhackenbeginnen(spiel Hack-Geräusche bis HHEnde) sauber in jede exit()-ähnliche Sache(CheatErkannt/Quit/Weltwechsel) einpflegen.

ThreadPerClient zu Threadpool ist wie C++ zu C.
(Türlich soll man bei einfachen Aufgaben das einfache Werkzeug benutzen, die Lib ist ja umschaltbar zwischen ThreadPerClient und Threadpool und in C++ muss man um 3 Zahlen zu sortieren keine Klassen benutzen.)

Könnte mir vorstellen, daß sauviele Cheats mit RAII auf dem Server gar nicht möglich wären.

Wenn mir die 30000 Threads ausgehen, hab ich ja schon 30000 Spieler und ausgesorgt. Im nächsten Spiel plane ich voraus und die Spielerthreads greifen nicht direkt mit mutex/lock->map auf die Weltdaten zu, sondern kommunizieren per Socket/Pipe mit dem Weltserver (fühlt sich aber genauso nach mutex/lock->map an). Und wenns dann wieder mehr als 30000 Kunden gibt, wird umgeschaltet auf Proxyrechner vor dem Weltrechner. 30000 Proxys vor dem Weltrechner, die je 30000 Spieler packen, das reicht erstmal.

hustbaer

Ich will auch mal ØMQ in die Runde geworfen haben:
http://zeromq.org/
WEIT davon entfernt eine Magic Bullet für dein Problem zu sein, aber definitiv eine sehr interessante Library.

Kellerautomat

Ich hab gehoert, dass zeromq keine ordentlich cancelbare IO hat oder so aehnlich. Irgendwas war damit, was die Lib so ziemlich unbrauchbar macht.