Re: pasynOctetSyncIO->read() weirdly slow performance

[Marco Filho via Tech-talk <tech-talk at aps.anl.gov>]

Hi! I had some interesting results! Turns out it was not asyn's read function the main responsible for the performance drag - although it has some considerable overhead - but epicsEventTryWait(stopEventId_) that was in the end of the loop.

A summary of the tests: all of them were made with 2232143 packets of 16 bytes each. The most basic client I was able to make in C++ could consume them in 0.6 seconds.

First test: using a raw socket, checking if epicsEvent stopEventId_ was sent. The code looks like appendix A. Took 105.8 seconds.
Second test: using a raw socket, no checking for the event. The code looks like appendix B. Obviously breaks the IOC but was only for the test. Took 0.665 seconds.

Third test: using asyn read function. Checking for the event. No explicit thread locking. The code looks like appendix C. Took 131.1 seconds.
Fourth test: using asyn read, not checking for event. No explicit thread locking. Code looks like appendix D. Took 24.1 seconds.
Fifth test: using asyn read, not checking for event, explicit thread locking. Code looks like appendix E. Took 1.134 second.

I was also curious about the overhead added by functions such as doCallbacksGenericPointer() and all other function from the AD framework. Although I have no good documented tests for now, they didn't add much overhead to the process. If I remember correctly, a naive code that read the acquisition, allocated a buffer, passed it as callback and released the buffer could read all packets in about 2 or 3 seconds.

In this particular case we will check some other methods of checking the signal to stop. I figure it would be useful to report since if I'm not mistaken this approach is used in a few AD modules.

Cheers and thanks for the help :)

Marco

Appendix A
#################################
 while(true){

if (!acquire) {
    asynPrint(this->pasynUserSelf, ASYN_TRACE_FLOW,
        "%s:%s: waiting for acquire to start\n", driverName, __FUNCTION__);

    epicsEventWait(startEventId_);
    /* Set needed parameters for acquisition */
    acquire = 1;
    this->lock();
    setIntegerParam(ADAcquire, acquire);
    setStringParam(ADStatusMessage, "Acquiring data");
    setIntegerParam(ADStatus, ADStatusAcquire);
    setIntegerParam(ADNumImagesCounter, 0);
    callParamCallbacks();
    this->unlock();

    /* Get needed parameters for acquisition */
    getIntegerParam(ADImageMode, &imageMode);
    getDoubleParam(readTimeout, &timeout);
    getIntegerParam(NDArrayCallbacks, &arrayCallbacks);

    /* Connect to ngem to receive images */
    reader.connectDevice();

}

/* Read NGEM data */
bytesTransferred = this->reader.recvData(tcp_frame, TCP_FRAME_SIZE);
//this->readData(eventType, arrayCallbacks, timeout);

if(epicsEventTryWait(stopEventId_) == epicsEventWaitOK) {
    reader.disconnectDevice();
    this->lock();
    if (imageMode == ADImageContinuous) {
        setIntegerParam(ADStatus, ADStatusIdle);
    } else {
        setIntegerParam(ADStatus, ADStatusAborted);
    }
    setIntegerParam(ADAcquire, 0);
    this->unlock();
    callParamCallbacks();
    memset(this->image, 0, sizeof(this->image));
    acquire = 0;
}
}
#################################

Appendix B
#################################
while (true) {

if (!acquire) {
    asynPrint(this->pasynUserSelf, ASYN_TRACE_FLOW,
        "%s:%s: waiting for acquire to start\n", driverName, __FUNCTION__);

    epicsEventWait(startEventId_);
    /* Set needed parameters for acquisition */
    acquire = 1;
    this->lock();
    setIntegerParam(ADAcquire, acquire);
    setStringParam(ADStatusMessage, "Acquiring data");
    setIntegerParam(ADStatus, ADStatusAcquire);
    setIntegerParam(ADNumImagesCounter, 0);
    callParamCallbacks();
    this->unlock();

    /* Get needed parameters for acquisition */
    getIntegerParam(ADImageMode, &imageMode);
    getDoubleParam(readTimeout, &timeout);
    getIntegerParam(NDArrayCallbacks, &arrayCallbacks);

    /* Connect to ngem to receive images */
    reader.connectDevice();

}

/* Read NGEM data */
bytesTransferred = this->reader.recvData(tcp_frame, TCP_FRAME_SIZE);
}

#################################
Appendix C
#################################

    while (1) {
        if (!acquire) {
            asynPrint(this->pasynUserSelf, ASYN_TRACE_FLOW,
                "%s:%s: waiting for acquire to start\n", driverName, __FUNCTION__);

            epicsEventWait(startEventId_);
            /* Set needed parameters for acquisition */
            acquire = 1;
            this->lock();
            setIntegerParam(ADAcquire, acquire);
            setStringParam(ADStatusMessage, "Acquiring data");
            setIntegerParam(ADStatus, ADStatusAcquire);
            setIntegerParam(ADNumImagesCounter, 0);
            callParamCallbacks();
            this->unlock();

            /* Get needed parameters for acquisition */
            getIntegerParam(ADImageMode, &imageMode);
            getDoubleParam(readTimeout, &timeout);
            getIntegerParam(NDArrayCallbacks, &arrayCallbacks);

            /* Connect to ngem to receive images */
            pasynCommonSyncIO->connectDevice(pasynUserCommon);

        }

        /* Read NGEM data */

        status = pasynOctetSyncIO->read(pasynUserOctet, (char*)tcp_frame, (size_t)TCP_FRAME_SIZE,
                                timeout, &nIn, &eomReason);

        if(epicsEventTryWait(stopEventId_) == epicsEventWaitOK) {
            pasynCommonSyncIO->disconnectDevice(pasynUserCommon);
            this->lock();
            if (imageMode == ADImageContinuous) {
                setIntegerParam(ADStatus, ADStatusIdle);
            } else {
                setIntegerParam(ADStatus, ADStatusAborted);
            }
            setIntegerParam(ADAcquire, 0);
            this->unlock();
            callParamCallbacks();
            memset(this->image, 0, sizeof(this->image));
            acquire = 0;
        }

    }

#####################
Appendix D
#####################

    while (1) {
        if (!acquire) {
            asynPrint(this->pasynUserSelf, ASYN_TRACE_FLOW,
                "%s:%s: waiting for acquire to start\n", driverName, __FUNCTION__);

            epicsEventWait(startEventId_);
            /* Set needed parameters for acquisition */
            acquire = 1;
            this->lock();
            setIntegerParam(ADAcquire, acquire);
            setStringParam(ADStatusMessage, "Acquiring data");
            setIntegerParam(ADStatus, ADStatusAcquire);
            setIntegerParam(ADNumImagesCounter, 0);
            callParamCallbacks();
            this->unlock();

            /* Get needed parameters for acquisition */
            getIntegerParam(ADImageMode, &imageMode);
            getDoubleParam(readTimeout, &timeout);
            getIntegerParam(NDArrayCallbacks, &arrayCallbacks);

            /* Connect to ngem to receive images */
            pasynCommonSyncIO->connectDevice(pasynUserCommon);

        }

        /* Read NGEM data */

        status = pasynOctetSyncIO->read(pasynUserOctet, (char*)tcp_frame, (size_t)TCP_FRAME_SIZE,
                                timeout, &nIn, &eomReason);

    }

###################
appendix E
###################

    while (1) {
        if (!acquire) {
            asynPrint(this->pasynUserSelf, ASYN_TRACE_FLOW,
                "%s:%s: waiting for acquire to start\n", driverName, __FUNCTION__);

            epicsEventWait(startEventId_);
            /* Set needed parameters for acquisition */
            acquire = 1;
            this->lock();
            setIntegerParam(ADAcquire, acquire);
            setStringParam(ADStatusMessage, "Acquiring data");
            setIntegerParam(ADStatus, ADStatusAcquire);
            setIntegerParam(ADNumImagesCounter, 0);
            callParamCallbacks();
            this->unlock();

            /* Get needed parameters for acquisition */
            getIntegerParam(ADImageMode, &imageMode);
            getDoubleParam(readTimeout, &timeout);
            getIntegerParam(NDArrayCallbacks, &arrayCallbacks);

            /* Connect to ngem to receive images */

            pasynCommonSyncIO->connectDevice(pasynUserCommon);

        }

        /* Read NGEM data */
        //this->readData(eventType, arrayCallbacks, timeout);
            struct octetPvt {
   asynCommon   *pasynCommon;
   void         *pcommonPvt;
   asynOctet    *pasynOctet;
   void         *octetPvt;
   asynDrvUser  *pasynDrvUser;
   void         *drvUserPvt;
    };
        octetPvt *pPvt = (octetPvt *) pasynUserOctet->userPvt;
        pasynUserOctet->timeout = timeout;
        pasynManager->lockPort(pasynUserOctet);
        status = pPvt->pasynOctet->read(pPvt->octetPvt, pasynUserOctet, (char*)tcp_frame, (size_t)TCP_FRAME_SIZE,
                                 &nIn, &eomReason);
        pasynManager->unlockPort(pasynUserOctet);


    }

---

From: Tech-talk <tech-talk-bounces at aps.anl.gov> on behalf of Marco Filho via Tech-talk <tech-talk at aps.anl.gov>
Sent: 29 October 2024 17:28:21
To: Mark Rivers; tech-talk at aps.anl.gov
Subject: Re: pasynOctetSyncIO->read() weirdly slow performance

 

Hi, Mark!

Thanks for the help.
I have just finished changing the driver for an experimental version which uses raw sockets.

However, only this by itself did not improve the performance as much as I intended.

I will compile some more organized results and share them as soon as I have something that makes sense.

Thanks so much for the help :)

Marco

---

From: Mark Rivers <rivers at cars.uchicago.edu>
Sent: 29 October 2024 17:07:11
To: Marco Filho; tech-talk at aps.anl.gov
Subject: RE: pasynOctetSyncIO->read() weirdly slow performance

 

Hi Marco,

 

If I understand correctly, your acquisition thread is calling pasynOctetSyncIO->read() 2,232,143 times in 24 seconds.  That is 10.1 microseconds per call.  If that is correct, then I am not surprised.  pasynOctetSyncIO->read calls pasynManager->queueLockPort() to gain exclusive access to the asyn port.  That call involves a context switch to a waiting thread.

 

A little history here.  Prior to asyn R4-14 in 2010 all the asynXXXSyncIO functions (including pasynOctetSyncIO->read) called pasynManager->lockPort() and pasynManager->unlockPort().  This is fast, but it causes problems as explained in the R4-14 release notes which say:

 

The problem with the previous implementation was that pasynManager->lockPort() immediately took the port mutex when it was available, rather than queueing a request to take the mutex. This could lead to one thread effectively getting exclusive access to the port, even if other threads had queued requests or tried to do SyncIO calls themselves. For example, if a device could send unsolicited input, then one might create a thread that simply called pasynOctetSyncIO->read() with a short timeout in a tight loop. The problem with this was that as soon as that thread released the port mutex when the read timed out, it would take the mutex again, blocking other threads that were trying to access the port. Previously the only solution to this problem was to add a short epicsThreadSleep() in the read loop.

 

The solution was to create a queueLockPort() function which queues a request to lock the port, allowing multiple threads to get access.  The penalty is that it is less efficient.

 

In your case where performance matters you really should not be calling pasynOctetSyncIO->read().  I suggest you change your acquisition thread to directly call pasynManager->lockPort(), pasynOctet->read(), and pasynManager->unlockPort().  To call pasynOctet->read() you need access to pasynOctet, and to the driver private handle.  You can do that by directly calling the pasynManager functions to find the interface, or you can cheat and take advantage of the fact that the pasynUser you are using already has that information, which was created in pasynUser->drvUser when pasynOctetSyncIO->connect() was called .  It is cheating because the structure definition is not publicly exposed, but perhaps it should be.  Here is some untested code:

 

struct octetPvt {

   asynCommon   *pasynCommon;

   void         *pcommonPvt;

   asynOctet    *pasynOctet;

   void         *octetPvt;

   asynDrvUser  *pasynDrvUser;

   void         *drvUserPvt;

};

 

octetPvt *pPvt = (octetPvt *) pasynUserOctet->userPvt;

pasynUserOctet->timeout = timeout;

pasynManager->lockPort(pasynUserOctet);

status = pPvt->pasynOctet->read(pPvt->octetPvt, pasynUserOctet, (char*)tcp_frame, (size_t)TCP_FRAME_SIZE,

                                &nIn, &eomReason);

pasynManager->unlockPort(pasynUserOctet);

 

Let me know if this works, and if it improves the performance.

 

Mark

 

 

From: Tech-talk <tech-talk-bounces at aps.anl.gov> On Behalf Of Marco Filho via Tech-talk
Sent: Tuesday, October 29, 2024 7:51 AM
To: tech-talk at aps.anl.gov
Subject: pasynOctetSyncIO->read() weirdly slow performance

 

Hi all, we are having a performance issue with asyn and I am not sure if I am misusing the tool...

So, we have a driver connected to a detector using a 1 Gb/s NIC. To simulate the detector behavior, we created a simple c++ code that opens a socket and generates 1 Gb of packets of 128 bits each. It then sends these packets to the loopback interface.
A simple c++ socket client could read these packets in a bit less than one second with no problem.
(Considering about 40 bytes of TCP+IP overhead, 1Gb of packets would be: (1*10^9)/(128 + 40*8) ≃ 2232143 packets)

We then tried to read from the loopback interface with the driver. It took about 140 seconds to read it! Thats about 7 Mbps or less than 1 MBps if I'm not making any silly mistake.
I then removed everything I could from the driver and left only the read function inside a while(true) loop. The read time reduced to 24 seconds. Better, but still weirdly slow.

Am I doing something wrong or is this asyn read function overhead expected?

About the code:

In the constructor, we connect to the detector via:

drvAsynIPPortConfigure(detectorPortName, detectorIPcmd.c_str(), 0, 1, 0);

 // /* Assign that port to the asynUser */
 pasynCommonSyncIO->connect(detectorPortName, 0, &pasynUserCommon, "");
 pasynOctetSyncIO->connect(detectorPortName, 0, &pasynUserOctet, "");

Then create the acquisition thread:

    epicsThreadOpts acqTaskThreadOpts;
    acqTaskThreadOpts.priority=epicsThreadPriorityMedium;
    acqTaskThreadOpts.stackSize=epicsThreadGetStackSize(epicsThreadStackMedium);
    acqTaskThreadOpts.joinable=1;
    status = (epicsThreadCreateOpt("acqTask",
                                    (EPICSTHREADFUNC)acqTaskC,
                                    this,
                                    &acqTaskThreadOpts) == NULL);

 

And inside the acquisition thread:

unsigned char tcp_frame[TCP_FRAME_SIZE];

    status = pasynOctetSyncIO->read(pasynUserOctet, (char*)tcp_frame, (size_t)TCP_FRAME_SIZE,
                                timeout, &nIn, &eomReason);

 

I also tried improving the thread priority from medium to maximum, but to no avail.

Thanks for any help,

Marco