Thanks for the detailed report.

> connections.py from multiprocessing has been rewrittend between 3.2
> and 3.3 but I can't see what's wrong in the way it has been done,
> basic socket options seem to be exactly the same.

Indeed, multiprocessing.connection has been rewritten in Python (by Antoine :-).

> One interesting point is that it seems to only affect the last bytes
> sent through the socket, e.g if you send a numpy.array big enough to
> fill the socket's read() buffer the first calls to read() are done at
> a normal speed, only the last one takes time.

And that's the catch.

Before, connection.send looked like this (for sockets):
"""
static Py_ssize_t
conn_send_string(ConnectionObject *conn, char *string, size_t length)
{
    Py_ssize_t res;
    /* The "header" of the message is a 32 bit unsigned number (in
       network order) which specifies the length of the "body".  If
       the message is shorter than about 16kb then it is quicker to
       combine the "header" and the "body" of the message and send
       them at once. */
    if (length < (16*1024)) {
        char *message;

        message = PyMem_Malloc(length+4);
        if (message == NULL)
            return MP_MEMORY_ERROR;

        *(UINT32*)message = htonl((UINT32)length);
        memcpy(message+4, string, length);
        Py_BEGIN_ALLOW_THREADS
        res = _conn_sendall(conn->handle, message, length+4);
        Py_END_ALLOW_THREADS
        PyMem_Free(message);
    } else {
        UINT32 lenbuff;

        if (length > MAX_MESSAGE_LENGTH)
            return MP_BAD_MESSAGE_LENGTH;

        lenbuff = htonl((UINT32)length);
        Py_BEGIN_ALLOW_THREADS
        res = _conn_sendall(conn->handle, (char*)&lenbuff, 4) ||
            _conn_sendall(conn->handle, string, length);
        Py_END_ALLOW_THREADS
    }
    return res;
}
"""

So, for short messages, the header was combined with the payload, and written at once to the socket, but for large messages, it was written in two steps.

Now the code looks like this:
"""
    def _send_bytes(self, buf):
        # For wire compatibility with 3.2 and lower
        n = len(buf)
        self._send(struct.pack("!i", n))
        # The condition is necessary to avoid "broken pipe" errors
        # when sending a 0-length buffer if the other end closed the pipe.
        if n > 0:
            self._send(buf)
"""

First the header is sent, then the payload.

The problem is that this kind of pattern causes problem because of a nasty interaction between Nagle's algorithm and delayed ack: in short, if the payload isn't large enough, the TCP stack won't send it before a short delay, see http://en.wikipedia.org/wiki/Nagle's_algorithm for more details.

So, one possible fix is to always combine the payload and header.
With the patch attached, we go from:
"""
$ ./python /tmp/test_manager.py
0.28662800788879395
0.08182215690612793
0.08193612098693848
0.08193612098693848
0.08194088935852051
0.08194208145141602
0.0819399356842041
0.08194184303283691
0.08194303512573242
0.0819389820098877
10
"""

to

"""
$ ./python /tmp/test_manager.py
0.04239797592163086
0.00041413307189941406
0.0004057884216308594
0.0004088878631591797
0.0004029273986816406
0.00040793418884277344
0.0004069805145263672
0.0004069805145263672
0.0004050731658935547
0.0004069805145263672
10
"""

Another possibility is to disable Nagle with TCP_NODELAY if available, or use some heuristic similar to the one in 3.2.