Node.js + socket.io, Cluster / Multithreading

[j4nk3y]

Guten Morgen zusammen,

Ich habe gestern das Cluster Modul von Node.js gefunden und würde dies gern in mein bestehendes Projekt implementieren. Dabei treten aber einige Probleme auf und ich hoffe jemand kann mir weiterhelfen.

Etwa folgendes läuft seit geraumer Zeit auf meinem Raspi problemlos:

var port = process.env.PORT || 1337;
var server = require('http').createServer().listen( port );
var io = require('socket.io').listen( server );

var user = require('./functions/server/js/user'); // Beispiel für ein eigens externes Modul.

io.sockets.on('connection', function( socket ){
	
	socket.on('getConnection', function( req ) {
		user.getConnection( socket, req, function ( res ) {
			socket.emit('setConnection', res);
		});
	});
  
  //.
  //.
  //.
});

Wenn ich das Beispiel folgenderweise bei mir umsetze, bekomme ich zum Beispiel, gar keine Verbindung mehr zum Server und erhalte im Browser nur ein "400 (Bad Reaquest)" beim Verbindundungsversuch zurück.

var cluster = require('cluster');

var port = process.env.PORT || 1337;
var threads = require('os').cpus().length;

var user = require('./functions/server/js/user');

if( cluster.isMaster ) {
	
	for(var i = 0; i < threads; i++) {
        cluster.fork();
    }
	
	cluster.on('online', function( thread ) {});
	
  cluster.on('exit', function( thread, code, signal) {
        cluster.fork();
  });
	
} else {
	var server = require('http').createServer().listen( port );
	var io = require('socket.io').listen( server );

	io.sockets.on('connection', function( socket ){

	  socket.on('getConnection', function( req ) {
			user.getConnection( socket, req, function ( res ) {
				socket.emit('setConnection', res);
			});
		});
  
  //.
  //.
  //.
});

Weiterhin werden alle externen Module für jeden thread neu geladen. Sprich in dem Beispiel wird var user = require('./functions/server/js/user'); bei mir 4 mal aufgerufen. Da ich vermute, dass die ganzen Daten im RAM gespeichert werden, würde ich gern die Beanspruchung so gering wie Möglich halten. Daher: Kann man das irgendwie verhindern, beziehungsweise user so Global setzen, dass jeder thread darauf zugreifen kann?

Schonmal vielen Dank für eure Hilfe.

Gruß
Jo

[1unitedpower]

Tipp ins Blaue: Deine Umgebungsvariablen werden nicht an die Kind-Prozesse weitergegeben, deswegen lauschen die Http-Server auf den Port 1337 und nicht auf process.env.PORT. Lösen könntest du das, indem du beim Forken die Umgebungsvariablen manuell durchreichst: cluster.fork(process.env).

Kann man das irgendwie verhindern, beziehungsweise user so Global setzen, dass jeder thread darauf zugreifen kann?

Jein, Threads haben in der Regel einen gemeinsamen Speicherbereich, aber cluster.fork erzeugt ganz neue Prozesse, die haben völlig getrennte Speicherbereiche. Du könntest aber die Arbeit die im User-Modul stattfindet nur im Master-Prozess erledigen lassen, und die Worker über die Ergebnisse per worker.send informieren. Das ist aber ganz schönes Gefrickel und du verlierst einige Vorteile der Parallelisierung, weil die Synchronisations-Punkte zwischen deinen Prozessen einbaust.

[dedlfix]

Tach!

Tipp ins Blaue: Deine Umgebungsvariablen werden nicht an die Kind-Prozesse weitergegeben, deswegen lauschen die Http-Server auf den Port 1337 und nicht auf process.env.PORT.

Das könnte man mit netstat überprüfen. Je nach Betriebssystem braucht man da den passenden Parameter, um sich lauschende Ports anzeigen zu lassen.

dedlfix.

[j4nk3y]

Moin,

Das könnte man mit netstat überprüfen. Je nach Betriebssystem braucht man da den passenden Parameter, um sich lauschende Ports anzeigen zu lassen.

netstat -l zeigt, dass Node auf 1337 lauscht.

Gruß
Jo

[j4nk3y]

Moin,

Tipp ins Blaue: Deine Umgebungsvariablen werden nicht an die Kind-Prozesse weitergegeben, deswegen lauschen die Http-Server auf den Port 1337 und nicht auf process.env.PORT. Lösen könntest du das, indem du beim Forken die Umgebungsvariablen manuell durchreichst: cluster.fork(process.env).

Hm... Ne funktioniert leider nicht, var port ist aber auch 1337 und nicht process.env.port, Müsste ich eh rausnehmen, da der Client sich zu Port 1337 verbinden möchte. Wie dem auch sei, habe ich mir gerade mal für beide Fälle das Server-Object per Console ausgeben lassen. Mit einem interessantem Unterschied:

Server {
  domain: null,
  _events:
   { connection: [Function: connectionListener],
     listening: [ [Function], [Function: bound ] ],
     close: [Function: bound ],
     upgrade: [Function],
     request: [Function] },
  _eventsCount: 5,
  _maxListeners: undefined,
  _connections: 0,
  _handle: null,
  _usingWorkers: false,
  _workers: [],
  _unref: false,
  allowHalfOpen: true,
  pauseOnConnect: false,
  httpAllowHalfOpen: false,
  timeout: 120000,
  keepAliveTimeout: 5000,
  _pendingResponseData: 0,
  maxHeadersCount: null,
  [Symbol(asyncId)]: -1 }

Das kommt bei der Cluster Methode.

Und bei der alten Methode zusätzlich diese Zeile: _connectionKey: '6::::1337', und danach [Symbol(asyncId)]: 6 }

Interessant aber viel kann ich mit der Information gerade nicht anfangen.

Jein, Threads haben in der Regel einen gemeinsamen Speicherbereich, aber cluster.fork erzeugt ganz neue Prozesse, die haben völlig getrennte Speicherbereiche. Du könntest aber die Arbeit die im User-Modul stattfindet nur im Master-Prozess erledigen lassen, und die Worker über die Ergebnisse per worker.send informieren. Das ist aber ganz schönes Gefrickel und du verlierst einige Vorteile der Parallelisierung, weil die Synchronisations-Punkte zwischen deinen Prozessen einbaust.

Ja, das habe ich mich auch schon fast gedacht. Das wäre denke ich zu kompliziert nachträglich noch zu realisieren.

Hm, ganz unterschiedliche Prozesse ist Schlecht... Obwohl kann mir egal sein, solange die wichtigen Daten richtig geschickt werden. Das sollte ich leicht hinkriegen.

Gruß
Jo

[j4nk3y]

Guten Morgen zusammen,

Folgendes hab ich jetzt mithilfe von diversen Modulen zusammen gestellt, was denke ich funktioniert.

Verwendet, bzw gerade installiert:
sticky-session
socket.io-redis brauch ich das??

//var cluster = require('cluster');
var http = require('http');
var threads = require('os').cpus().length;
//var redis = require('socket.io-redis');

var sticky = require('sticky-session');
var port = 1337;
var server = http.createServer()
var io = require('socket.io').listen( server );

var user = require('./functions/server/js/user');

if( !sticky.listen( server, port ) ) {
	
	console.log( '[ Node ] Server running on Port : ' + port );
	console.log( '[ Node ] Master cluster setting up ' + threads + ' threads' );
	cluster.on('online', function( thread ) {
       console.log( '[ Node ] Cluster thread ' + cluster.worker.id + ' set up' );
   });
	
   cluster.on( 'exit', function( thread, code, signal) {
       console.log( '[ Node ] Cluster thread ' + thread.id + ' died with code: ' + code + ', >and signal: ' + signal );
   });
} else {

	io.sockets.on('connection', function( socket ){

	  socket.on('getConnection', function( req ) {
			user.getConnection( socket, req, function ( res ) {
				socket.emit('setConnection', res);
			});
		});
  
  //.
  //.
  //.
});

Ist das richtig?

Also ich bekomme eine Verbindung und es werden 4 Prozesse initialisiert.
Einige kleine Tests ergeben auch das unterschiedliche Clients auf den unterschiedlichen 'workern' landen und das ein 'worker' bei einem unerwartetem 'exit' neu gestartet wird.

Wie das so ist, folgt nach einem Problem gleich das nächste. Dafür habe ich eigentlich eine Lösung, persönlich gefällt mir die aber nicht.
Ich habe ein Globales Object mit var global = require('./functions/server/js/global'); eingebunden, welches ein Abbild von (größtenteils) statischen Werten aus meiner Datenbank ist. Diese Werte nutze ich zum Beispiel in Berechnungen. Außerdem nutzen diverse Cron-jobs diese Daten um einige Werte Zeitabhängig neu zu berechnen. Diese Lösung finde ich sehr praktisch, da die Cron-jobs sowie durch einen Client getriggerte Berechnungen nicht erst die Datenbank abfragen müssen sondern direkt die Werte aus global.xyz nehmen können und die Ergebnisse in der Datenbank einfach aktualisiert werden können.
Damit nutze ich die Datenbank im Moment noch Primär als Backup, falls der Node Prozess ausfällt (und in global.js werden die Daten bei jedem Neustart des Prozesses aus der Datenbank geladen).

Lösung, alle (benötigten) Daten bei jedem Event (Client oder Cron getriggert) neu aus der Datenbank holen und dann aktualisieren. Wie gesagt, erscheint mir das nicht elegant, bietet aber den Vorteil, dass jeder 'worker' die gleichen Daten aus der Datenbank hat.

Was mich da etwas verunsichert ist, dass ich nicht weiß ob jeder 'worker' eine Referenz auf das Objekt bekommt wenn ich es zuerst im Master einbinde und dann in den einzelnen 'workern'.

Gruß
Jo

[j4nk3y]

Hey,

Ich habe ein Modul gefunden mit dem sich eine globale Schnittstelle von Daten bewerkstelligen lassen würde. Nur habe ich die Befürchtung, dass wenn 2 Prozesse gleiche Daten bearbeiten oder abfragen, dass dies zu Fehlern führen könnte.
Beliebtes Beispiel ist hier ja immer etwas wie, 2 Personen bestellen den letzten Apfel im Sortiment und beide bezahlen diesen nur derjenige, dessen Prozess 1 Millisekunde später fertig ist, bekommt keinen Apfel, da dieser dann schon nicht mehr im Sortiment ist.

Um so etwas zu verhindern, wäre die Datenbank variante besser, obwohl ich da nicht weiß ob ich in dem mysql-pool ein Transaction eingebaut bekomme, welches ja genau das verhindern soll.

Gruß
Jo