node.js socket exception read ETIMEDOUT - how do I catch it properly? what about write timeouts?

Question

I have a proxy server that manages a bunch of clients and also talks with another http server. Messages get posted back and forth and directed to clients. Clients can and do timeout, and the server has a heartbeat function (that repeats every n seconds) that sends a heartbeat to all clients that are in a map of clientId to socket connection.

I get a 'read ETIMEDOUT' exception when the heartbeat tries to talk to a client that is no longer connected but who's socket is still active. I tried temporarily setting the timeout of the socket connection to 2000ms with the theory that my socket event handler for timeout would catch this (the event handler is in the tcp server portion), but this didn't happen. It takes several heartbeats to die.

Part of the problem is definitely my lack of understanding of how to structure node.js code, so if you have any suggestions, I'd very much appreciate them.

Another question is whether it is possible to handle read and write timeouts separately, or at least break them out. What I'd really like to do is have my heartbeat function be part of the tcp server and only send a heartbeat if it hasn't heard from the client in say n seconds, and only send this heartbeat once. If we get a timeout, then we kill the socket, otherwise we wait again.

Thanks!

>>$ node --harmony-weakmaps server.js
Heartbeat: Sat Feb 18 2012 08:34:40 GMT+0000 (UTC)
{
    sending keep_alive to id:00:00:00:00:00:10 socket:[object Object]
}
socket:received data: {"id":"00:00:00:00:00:10","m":"keep_alive","success":"true"}

Heartbeat: Sat Feb 18 2012 08:35:40 GMT+0000 (UTC)
{
    sending keep_alive to id:00:00:00:00:00:10 socket:[object Object]
}


socket:received data: {"id":"00:00:00:00:00:10","m":"keep_alive","success":"true"}
node.js:201
        throw e; // process.nextTick error, or 'error' event on first tick
              ^
Error: read ETIMEDOUT
    at errnoException (net.js:642:11)
    at TCP.onread (net.js:375:20)

Heartbeat function that triggers the timeout:

console.log("Starting heartbeat");
var beat_period = 60;
setInterval(function() {
    if(Object.keys(id2socket).length != 0){
        console.log("Heartbeat: " + new Date());
        //for (var key in id2socket) {
        //  console.log("\t"+key+"->"+id2socket[key]);
        //}
        console.log("{");
        for(var id in id2socket) {
            var socket = id2socket[id];
            // don't want sockets to time out
            socket.setTimeout(2000); // for heartbeat, set the timeout
            try {
                console.log("\tsending keep_alive to id:"+id+" socket:"+id2socket[id]);
                socket.write('{"m":"keep_alive"}\r\n');
            } catch(Error) {
                console.log("Heartbeat:Cannot find id:"+id);
                removeSocketFromMap(id,socket);
                // TODO: send message to API
            }
            socket.setTimeout(0); // no timeout
        }
        console.log("}");
    }
}, beat_period * 1000);

server.js:

// Launch Instructions
// node --harmony-weakmaps server.js

var net = require('net'); // tcp-server
var http = require("http"); // http-server
var querystring = require('querystring');

// Map of sockets to clients
var id2socket = new Object;
var socket2id = new WeakMap; // allows us to use object as key to hash

// Test for client:
// {"id":"123","m":"add"} // establishes connection and puts client into id2socket map
// {"id":"123","m":"test"} // sends a message through to API

// HTTP:POST outbound function
// http://www.theroamingcoder.com/node/111
function postOut(dataToPost){
    try{
        console.log("postOut msg:"+JSON.stringify(dataToPost));
    } catch (Error) {
        console.log("postOut error:"+Error);
    }

    var post_domain = '127.0.0.1';
    var post_port = 80;
    var post_path = '/cgi-bin/index3.py'; 

    var post_data = querystring.stringify({
            'act' : 'testjson',
            'json' : JSON.stringify(dataToPost)
    });
    console.log("post_data:"+post_data);

    var post_options = {
      host: post_domain,
      port: post_port,
      path: post_path,
      method: 'POST',
      headers: {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Content-Length': post_data.length
      }
    };
    var post_req = http.request(post_options, function(res) {
      res.setEncoding('utf8');
      res.on('data', function (chunk) {
        console.log('Response:data: ' + chunk);
      });
    });

    // Handle various issues
    post_req.on('error', function(error) {
        console.log('ERROR' + error.message);
        // If you need to go on even if there is an error add below line
        //getSomething(i + 1);
    });
    post_req.on("response", function (response) {
        console.log("Response:response:"+response);
    });

    // write parameters to post body
    post_req.write(post_data);
    post_req.end();
}

function removeSocketFromMap(id,socket){
    console.log("removeSocketFromMap socket:"+socket+" id:"+id);
    delete id2socket[id];
    socket2id.delete(socket);
    //TODO: print map???
    console.log("socketmap {");
    for (var key in id2socket) {
        console.log("\t"+key+"->"+id2socket[key]);
    }
    console.log("}");
}

// Setup a tcp server
var server_plug = net.createServer(

    function(socket) {

        // Event handlers
        socket.addListener("connect", function(conn) {
            console.log("socket:connection from: " + socket.remoteAddress + ":" + socket.remotePort + " id:"+socket.id );   
        });

        socket.addListener("data", function(data) {
            console.log("socket:received data: " + data);
            var request = null;
            try {
                request = JSON.parse(data);
            } catch (SyntaxError) {
                console.log('Invalid JSON:' + data);
                socket.write('{"success":"false","response":"invalid JSON"}\r\n');
            }

            if(request!=null){
                response = request; // set up the response we send back to the client

                if(request.m=="keep_alive"){ // HACK for keep alive
                    // Do nothing
                } else if(request.m !== undefined && request['id'] !== undefined){ // hack on 'id', id is js obj property
                    if(request.m == 'connect_device' || request.m == 'add'){
                        console.log("associating uid " + request['id'] + " with socket " + socket);
                        id2socket[request['id']] = socket;
                        socket2id.set(socket, request['id']);
                    }
                    postOut(request);
                    socket.write(JSON.stringify(response)+"\r\n");
                } else if(request['id'] !== undefined){
                    postOut(request);
                    socket.write(JSON.stringify(response)+"\r\n");
                } else {
                    response['content'] = "JSON doesn't contain m or id params";
                    socket.write(JSON.stringify(response)+"\r\n");
                }
            } else {
                console.log("null request");
            }

        });

        socket.on('end', function() {
            id = socket2id.get(socket);

            console.log("socket:disconnect by id " + id);
            removeSocketFromMap(id,socket);
            socket.destroy();
        });

        socket.on('timeout', function() {
            id = socket2id.get(socket);

            console.log('socket:timeout by id ' + id);
            removeSocketFromMap(id,socket);
            socket.destroy();
        });

        // handle uncaught exceptions
        socket.on('uncaughtException', function(err) {
            id = socket2id.get(socket);

            console.log('socket:uncaughtException by id ' + id);
            removeSocketFromMap(id,socket);
            socket.destroy();
        });

    }
);
server_plug.on('error', function (error) {
    console.log('server_plug:Error: ' + error);
});

// Setup http server
var server_http = http.createServer(
    // Function to handle http:post requests, need two parts to it
    // http://jnjnjn.com/113/node-js-for-noobs-grabbing-post-content/
    function onRequest(request, response) {
        request.setEncoding("utf8");
        request.content = '';

        request.on('error', function(err){
            console.log("server_http:error: "+err);
        })

        request.addListener("data", function(chunk) {
            request.content += chunk;
        });

        request.addListener("end", function() {
            console.log("server_http:request_received");

            try {
                var json = querystring.parse(request.content);

                console.log("server_http:received_post {");
                for(var foo in json){
                    console.log("\t"+foo+"->"+json[foo]);
                }
                console.log("}");

                // Send json message content to socket
                if(json['json']!=null && json['id']!=null){
                    id = json['id'];
                    try {
                        var socket = id2socket[id];
                        socket.write(json['json']+"\r\n");
                    } catch (Error) {
                        console.log("Cannot find socket with id "+id);
                    } finally {
                        // respond to the incoming http request
                        response.end();
                        // TODO: This should really be in socket.read!
                    }
                }
            } catch(Error) {
                console.log("JSON parse error: "+Error)
            }
        });

        request.on('end', function () {
            console.log("http_request:end");
        });

        request.on('close', function () {
            console.log("http_request:close");
        });
    }
);
server_http.on('error', function (error) {
    console.log('server_http:Error: ' + error);
});

// Heartbeat function
console.log("Starting heartbeat");
var beat_period = 60;
setInterval(function() {
    if(Object.keys(id2socket).length != 0){
        console.log("Heartbeat: " + new Date());
        //for (var key in id2socket) {
        //  console.log("\t"+key+"->"+id2socket[key]);
        //}
        console.log("{");
        for(var id in id2socket) {
            var socket = id2socket[id];
            // don't want sockets to time out
            socket.setTimeout(2000); // for heartbeat, set the timeout
            try {
                console.log("\tsending keep_alive to id:"+id+" socket:"+id2socket[id]);
                socket.write('{"m":"keep_alive"}\r\n');
            } catch(Error) {
                console.log("Heartbeat:Cannot find id:"+id);
                removeSocketFromMap(id,socket);
                // TODO: send message to API
            }
            socket.setTimeout(0); // no timeout
        }
        console.log("}");
    }
}, beat_period * 1000);



// Fire up the servers
//var HOST = '127.0.0.1'; // just local incoming connections
var HOST = '0.0.0.0'; // allows access to all external IPs
var PORT = 5280;
var PORT2 = 9001;

// accept tcp-ip connections
server_plug.listen(PORT, HOST);
console.log("TCP server listening on "+HOST+":"+PORT);

// accept posts
server_http.listen(PORT2);
console.log("HTTP server listening on "+HOST+":"+PORT2);

EDIT:

Should I be using .on(event,callback) vs .onlistener(event,callback)?

UPDATE:

That didn't work, I changed the stuff in tcp_server to all add_listener in the heartbeat as .on. Still didn't catch the errors and blew up and said I added too many listeners.

Answer 1

First, its a bit hard to say if your structure is right without some more understanding of the context of your code...

Try adding

socket.on('error', function() {
    id = socket2id.get(socket);

    console.log('socket:timeout by id ' + id);
    removeSocketFromMap(id,socket);
    socket.destroy();
}

to the anonymous function in net.CreateServer. ETIMEDOUT is a system call error, and node.js is just reporting it. It may not be caused by just a typical 'timeout'. You say its being caused by the Hearbeat write, but it looks like TCP.read is the origination. It may be a half-closed socket.