How to get a stream to await for incoming data on a socket

[ejm4567]

Hi Everyone,

I get a EndOfStream raised exception when running the following code.
What I want is for the script to run continuously, waiting for incoming data on a socket.

Code snippet:

from anyio import (
       TASK_STATUS_IGNORED,
        create_task_group,
        connect_tcp,
        create_tcp_listener,
        run,
     )

from  anyio.abc import TaskStatus

from sums_classes import SUMS_Requests
...

async def sums_handler(stream): 
       sr = SUMS_Requests(stream, log)
       await sr.ahandle(stream)                           <--all the work is done in this method

async def sum_server(
      port: int, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED
      ):
      async with await create_tcp_listener(
              local_host="127.0.0.1", local_port=port
             ) as listener:
                task_status.started()
                await listener.serve(sums_handler)

async def sums_main():
     async with create_task_group as tg:
          await tg.start(sum_server, 6002)
          async with await connect_tcp("127.0.0.1", 6002) as stream:
                 log.writeInfo(['sums_main task group started...'])

try:
    run(sums_main)
except KeyboardInterrupt:
    raise KeyboardInterrupt


class SUMS_Requests(SUMSRequest):
     def __init__(self, stream, log):
           self.stream = stream
           self.log = log
           self.setup()                  <--method that does some initialization
          return None

  async def ahandle(self, stream):
           self.stream = stream

           while 1:
                  msg_len = await self.stream.receive(8)        <---read the first 8 bytes in the stream.  this is where EndOfStream is raised 

                 if len(msg_len) == 8:
                       break

         ....    lots more code here processing the 4 other messages that arrive from the server

Do I need to use a SocketListener, or raise a custom exception that is passed when no data is there.

The use case is this python script needs to run continuously waiting for messages coming from a server code, processing them, and sending the results back to the server, then wait until new messages come in again.

Thanks,

--Ed

[agronholm]

As the documentation explains, EndOfStream is raised when you try to receive data from the remote end but no data is available and the remote end has closed the connection. Therefore it doesn't make sense to try to do anything else with that stream. I should also point out that the comment implies a misunderstanding in the logic of the stream receive operation: receive(8) isn't guaranteed to read exactly 8 bytes, it reads at most 8 bytes. So you could get fewer too. To overcome this, wrap it in BufferedByteReceiveStream and call buffered_stream.receive_exactly(8).

[ejm4567]

I was just reading the documentation on BufferedByteReceiveStream. Let me test that out, it looks promising.
Thanks for the tip.

[ejm4567]

I think one issue is I need to wait for the socket to be readable.
Otherwise, as soon as I bring up the client it attempts to read from the socket and fails with an EndOfStream.

I think the answer is to use wait_socket_readable, and backend = 'trio', with code similar to GitHub issue #201 :

from anyio import (
       TASK_STATUS_IGNORED,
        create_task_group,
        connect_tcp,
        create_tcp_listener,
        run,
     )

from  anyio.abc import TaskStatus, SocketAttribute

from sums_classes import SUMS_Requests
...

async def sums_handler(stream): 
       sock = stream.extra(SocketAttribute.raw_socket)
       sr = SUMS_Requests(sock, log)
       await sr.ahandle(sock)                           <--all the work is done in this method, now passing an stdlib socket

async def sum_server(
      port: int, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED
      ):
      async with await create_tcp_listener(
              local_host="127.0.0.1", local_port=port
             ) as listener:
                task_status.started()
                await listener.serve(sums_handler)

async def sums_main():
     async with create_task_group as tg:
          await tg.start(sum_server, 6002)
          async with await connect_tcp("127.0.0.1", 6002) as stream:
                 log.writeInfo(['sums_main task group started...'])

try:
    run(sums_main, backend='trio')
except KeyboardInterrupt:
    raise KeyboardInterrupt

class SUMS_Requests(SUMSRequest):
     def __init__(self, sock, log):
           self.sock = sock
           self.log = log
           self.setup()                  <--method that does some initialization
          return None

  async def ahandle(self, sock):
           self.sock = sock
           self.conn = sock.fileno()

           while 1:
                 await wait_socket_readable(self.conn)                <--this command holds the client until the server sends a message
                  msg_len, aux, *_ = sock.revcmsg(8, 1024)

                 if len(msg_len) == 8:
                       break

                 msg_len_decoded = msg_len.decode('UTF-8')
                 msgLength = int(msg_len_decoded, 16)
      
           while 1:
                 await wait_socket_readable(self.conn)
                  self.msg, aux, *_ = sock.revcmsg(msgLength, 1024)

                 if len(self.msg) == msgLength:
                       break

                --code to convert a JSON message to a dict then call a factory to send back a JSON message
                                (converted back to a bytes object

              aliveRtn = await self.sendJson(aliveMsg)

     ...

class SUMRequest(Request):                    <--parent (base) class to SUMSRequests
  async def sendJson(self, msgStr):
             msgBytes = bytes(msgStr, 'UTF-8')
             bytesSent = await self.sendMsg(msgBytes)
             return bytesSent

  async def sendMsg(self, msg):
            numBytesMessage = '{:08x}', format(len(msg))
            self.log.writeInfo(['THERE...' + numBytesMessage])

           self.log.writeInfo(['Inside sendMsg  msg = %s.  msg_len = %d' % (msg, msg_len)])
           bytesSent = await self.sock.send(bytesarray(numBytesMessage]bytesSentTotal:], 'UTF-8'))   <--send the length of the message first, as a hex decimal

           bytesSent = await self.sock.send(msg)   <--now send the message
           return bytesSent
  

The server code that sends the initial messages is a big legacy C code, using the C socket library.
The client side is a python script (snippet above).

The result is the first 2 messages are received fine by the Python code.
When the reply is sent back to the server, the length of the message is sent first (as a hex decimal), then the message itself.

Apparently, the C code never gets the message from python, so throws an error "Socket to sumsd.py broken"

The log file from the server is:

[drms8@hydra-eikonal-net linux_x86_64]$ ./vso_sum_alloc sunum=1310756190 size=282822859 seriesname=aia.lev1
EJM---ConnectToMtSums  sending JSON...
EJM--- Buffer(msg length) = {
	"pid":	1888675,
	"user":	"drms8"
}
EJM--- Buffer(msg) = {
	"pid":	1888675,
	"user":	"drms8"
}   length of msg = 37
EJM---Sent buffer   bytesSent = 37
EJM---Calling MTSums...
EJM--- Buffer(msg length) = {
	"reqtype":	"open"
}
EJM--- Buffer(msg) = {
	"reqtype":	"open"
}   length of msg = 22
EJM---Sent buffer   bytesSent = 22
EJM---Calling receiveMsg...
EJM---SockFD = 3
EJM --nReady = 1
Socket to sumsd.py broken.               <----here is where the C code dies due to no data received
Segmentation fault (core dumped)
[drms8@hydra-eikonal-net linux_x86_64]

The corresponding log on the Python side is:

2023-11-27 09:03:58 - INFO - sumsd_classes5.py:3950: Inside SUMS_Requests constructor...
2023-11-27 09:03:58 - INFO - sumsd_classes5.py:560: user drms8 successfully connected to data_sums database: localhost:5470 - id 0
2023-11-27 09:03:58 - INFO - sumsd_classes5.py:4258: At top of ahandle...<class 'socket.socket'>
2023-11-27 09:03:58 - INFO - sumsd-jmd5.py:281: sums_main task group started...
2023-11-27 09:04:22 - INFO - sumsd_classes5.py:3950: Inside SUMS_Requests constructor...
2023-11-27 09:04:22 - INFO - sumsd_classes5.py:560: user drms8 successfully connected to data_sums database: localhost:5470 - id 1
2023-11-27 09:04:22 - INFO - sumsd_classes5.py:4258: At top of ahandle...<class 'socket.socket'>  <--this is type(sock) and seems to be the stdlib socket as expected
2023-11-27 09:04:22 - INFO - sumsd_classes5.py:4279: is-alive message received and decoded...
2023-11-27 09:04:22 - INFO - sumsd_classes5.py:4296: Inside handler  msgDict = {'pid': 1888675, 'user': 'drms8'}
2023-11-27 09:04:22 - INFO - sumsd_classes5.py:4311: HERE...
2023-11-27 09:04:22 - INFO - sumsd_classes5.py:3134: Inside sendJson  msg len = 63
2023-11-27 09:04:22 - INFO - sumsd_classes5.py:3102: Inside sendMsg...loop = <class 'socket.socket'>  <--same as above, now in sendMsg
2023-11-27 09:04:22 - INFO - sumsd_classes5.py:3105: THERE...0000003f 
2023-11-27 09:04:22 - INFO - sumsd_classes5.py:3111: Inside sendMsg  msg = b'{"status": "ok", "existing": "indeterminate", "is-alive": true}'   msg_len = 63
2023-11-27 09:04:22 - INFO - sumsd-jmd5.py:296: Inside finally block...

NOTE: We are using the trio backend, the same as in GitHub issue #201
When I tried asyncio as the backend, I get the same error message as seen in issue #201, even though I am using Anyio 4.0 and python 3.11.6

Is the stdlib socket in Trio identical to the underlying socket.socket built into Python itself?
Or is there an additional wrapper involved in Trio?

Likewise, does anyio simply pass along to the specified backend the socket ?
Or does anyio attach any additional wrapper function to the socket before passing it along to the backend?

I have a version of the above Python client script that uses socket.socket and asyncio directly, and that script is able to send messages, via sendall, back to the server.

[agronholm]

I think one issue is I need to wait for the socket to be readable.
Otherwise, as soon as I bring up the client it attempts to read from the socket and fails with an EndOfStream.

No, that's not how it works. The receive operation will block until there are bytes to be read, or the remote end closes the connection (in which case EndOfStream is raised).

Is the stdlib socket in Trio identical to the underlying socket.socket built into Python itself?
Or is there an additional wrapper involved in Trio?

Trio wraps the socket because socket.socket does not have any wait to asynchronously wait for operations to complete.

Likewise, does anyio simply pass along to the specified backend the socket ?
Or does anyio attach any additional wrapper function to the socket before passing it along to the backend?

AnyIO has a lightweight wrapper around the trio socket as an adapter between AnyIO's and Trio's APIs.

[ejm4567]

OK, I've resolved the issue of the Segmentation fault. The Segmentation fault was due to a bug in the legacy C code where the requests to the Python script originate. Fixing the bug in the legacy C code then allowed the Python script to run continuously when the socket is readable, and wait for an indefinite period (which can vary from a few seconds, to up to 10 mins) as the legacy codebase works in it's caching DB and queues.

For the record, here's the key snippet of AnyIO code that works with the legacy C code driver:

from anyio import (
        TASK_STATUS_IGNORED,
        create_task_group,
        connect_tcp,
        create_tcp_listener,
        wait_socket_readable,
        run,
        BrokenResourceError, ClosedResourceError, EndOfStream,
        )
from anyio.abc import TaskGroup, TaskStatus, SocketAttribute
    ....

#  Define the main handler sums_handler for requests from the C code

    async def sums_handler(stream):
        receive_sock = stream.extra(SocketAttribute.raw_socket)
        log.writeDebug(['Inside sums_handler   receive_sock = %s' % (str(type(receive_sock)))])
        log.writeDebug(['Inside sums_handler   receive_stream = %s' % (str(type(stream)))])
        sr = SUMS_Requests(stream, log, pdebug, metrics)
        await sr.ahandle(receive_sock)

#  Define the async function sum_server which creates a TCP listener, which in turn serves out sums_handler

    async def sum_server(port: int, host: str, *, task_status: TaskStatus[None] = TASK_STATUS_IGNORED):
            async with await create_tcp_listener(local_host=host, local_port=port, backlog=65536, reuse_port=True) as listener:
                task_status.started()
                await listener.serve(sums_handler)

#  Define the async function sums_main which creates the initial Task group, starts it, and then establishes a context manager "stream" via connect_tcp, to connect to sum_server

    async def sums_main():
        async with create_task_group() as tg:
            host = arguments.getArg('listenhost')
            port = arguments.getArg('listenport')
            await tg.start(sum_server, port, host)
            async with await connect_tcp(host, port) as stream:
                log.writeInfo(['sums_main task group started...'])

#  Main try/except block that runs sums_main using the AnyIO backend trio

    try:
        run(sums_main, backend='trio')
    except KeyboardInterrupt:
        log.writeInfo(['Received SIGINT from stop script.... shutting down now    pid = ' + str(pid)])
        raise KeyboardInterrupt
    except BrokenResourceError:
        raise BrokenResourceError
    except ClosedResourceError:
        raise ClosedResourceError
    except EndOfStream:
        raise EndOfStream
    except:
        log.writeInfo(['Caught an exception...'])
        (texc, excp, tb) = sys.exc_info()
        log.writeInfo(['Exception type = %s  name = %s  traceback = %s' % (str(texc), str(excp), str(tb))])
        pass
    else:
        log.writeInfo(['Inside else block...'])

    log.writeInfo(['Closing down SUMS server...'])

The main class SUMS_Requests snippet is:

################################################################
#  SUMS Requests class  --used by AnyIO task sum_server
################################################################

class SUMS_Requests(SUMSRequest):
    def __init__(self, stream, log, pdebug, metrics):
        self.stream = stream
        self.log = log
        self.pdebug = pdebug
        self.metrics = metrics

        self.setup()
        return None

   ....

#  Asynchrous TCP Handler code for AnyIO function sums_handler

    async def ahandle(self, sock):
        self.sock = sock
        self.conn = sock.fileno()
        if self.pdebug:
            self.log.writeInfo(['---------TOP--------'])
            self.log.writeInfo(['At top of ahandle...'+str(type(sock))])

#  is-alive msg section

        msgLength = await self.getLength(8)
        self.msg = await self.getMessage(msgLength)

        self.msgDict = Unjsonizer(self.msg)
        sumsFactory = SUMSRequestFactory(self, log=self.log, dbconn=self.dbconn, dbobj=self.dbobj, sconn=self.sock, pdebug=self.pdebug)

        if 'user' in self.msgDict.unjsonized:
            aliveRequest = sumsFactory.getRequest(self.msg)
            aliveMsg = aliveRequest.generateAliveResponse().getJSON()
            aliveRtn = self.sendJson(aliveMsg)
            if self.pdebug:
                self.log.writeInfo(['Inside handler  is-alive completed...  aliveMsg = ' + str(aliveMsg)])

     ...

     similar code blocks for the other 4 types of messages received from the C code

To simplify the ahandle method, two new methods are added, getLength and getMessage:

    async def getLength(self, buffer_len):

        while 1:
            try:
                await wait_socket_readable(self.conn)
            except:
                self.log.writeInfo(['Caught an exception while waiting for a readable socket...'])
                (texc, excp, tb) = sys.exc_info()
                self.log.writeInfo(['Exception type = %s  name = %s  traceback = %s' % (str(texc), str(excp), str(tb))])
                raise Exception
            try:
                msg_len, aux, *_ = self.sock.recvmsg(buffer_len, 1024)
                if len(msg_len) == buffer_len:
                    break
            except KeyboardInterrupt:
                raise KeyboardInterrupt
            except IncompleteRead:
                raise IncompleteRead
            except:
                self.log.writeInfo(['Caught an exception in getLength...'])
                (texc, excp, tb) = sys.exc_info()
                self.log.writeInfo(['Exception type = %s  name = %s  traceback = %s' % (str(texc), str(excp), str(tb))])
                pass

        msg_len_decoded = msg_len.decode('UTF-8')
        msgLength = int(msg_len_decoded,16)
        return msgLength

    async def getMessage(self, msgLength):

        while 1:
            await wait_socket_readable(self.conn)
            try:
                msg, aux, *_ = self.sock.recvmsg(msgLength, 1024)
                if len(msg) == msgLength:
                    break
            except IncompleteRead:
                pass
            except:
                self.log.writeInfo(['Caught an exception in getMessage...'])
                (texc, excp, tb) = sys.exc_info()
                self.log.writeInfo(['Exception type = %s  name = %s  traceback = %s' % (str(texc), str(excp), str(tb))])
                pass

        return msg

Three points should be emphasized:

1. The legacy C code binds to a socket, not a stream, so the python script needs access to the raw socket in order to call wait_socket_readable. Otherwise, if streams are used on the python side, the legacy C code simply never responds. Depending on the legacy code in the wild, this may well be a common occurrence, and may warrant some additional documentation in AnyIO.
2. The raw socket file descriptor needed by wait_socket_readable is obtained from SocketAttribute from a SocketStream.
3. I've only been able to get the above AnyIO code to work with the Trio backend. It fails using Asyncio. This may indicate that Github issue Using "recvmsg" with anyio #201 is still an issue.

Observation:

In testing SocketAttribute, I found that the example for typed attributes failed, specifically importing TypedAttribute failed. I don't see a class statement for TypedAttribute in the AnyIO module _core/_typedattr.py
Does the documentation for typed attribute need updating, or is this a bug?

[agronholm]

It looks like the documentation needs updating.

[agronholm]

I have a PR open here: #683
Would you mind having a quick look?