Queue optimizations (#29)

* QUEUE_NO_LOCK
* Respect flush requests when QUEUE_PEEK_THRESHOLD mode

Author: RainerZ

examples/tokio_demo/src/main.rs

@@ -93,7 +93,7 @@ const CALPAGE1: CalPage1 = CalPage1 {
 // Stop after 5s
 // Trigger a global event when any task starts or stops
 // Trigger a thread local event, in each loop
-// Once the A2L registry is created on XCP client connect, tli events and variable instances are fixed and addional instances are not visible
+// Once the A2L registry is created on XCP client connect, tli events and variable instances are fixed and additional instances are not visible
 // Tokio occasionally creates new worker threads and destroys old ones very late, so the number of instances may change
 
 #[allow(dead_code)]
@@ -227,7 +227,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
             counter = 0;
         }
 
-        // Triger the measurement event "task"
+        // Trigger the measurement event "task"
         // The measurement event timestamp is taken here and captured data is sent to CANape
         event.trigger();
     }

tests/test_multi_thread.rs

@@ -25,11 +25,11 @@ use xcp_test_executor::test_executor;
 const TEST_CAL: xcp_test_executor::TestModeCal = xcp_test_executor::TestModeCal::Cal; // Execute calibration tests: Cal or None
 const TEST_DAQ: xcp_test_executor::TestModeDaq = xcp_test_executor::TestModeDaq::DaqMultiThread; // Execute measurement tests: MultiThreadDAQ or None
 
-const TEST_TASK_COUNT: usize = 50; // Number of test tasks to create
+const TEST_TASK_COUNT: usize = 50; // Number of test tasks (threads) to create
 const TEST_SIGNAL_COUNT: usize = 32; // Number of signals is TEST_SIGNAL_COUNT + 5 for each task
 const TEST_DURATION_MS: u64 = 10 * 1000; // Stop after TEST_DURATION_MS milliseconds
-const TEST_CYCLE_TIME_US: u32 = 250; // Cycle time in microseconds
-const TEST_QUEUE_SIZE: u32 = 1024 * 1024; // Size of the XCP server transmit queue in Bytes
+const TEST_CYCLE_TIME_US: u32 = 100; // Cycle time in microseconds
+const TEST_QUEUE_SIZE: u32 = 2 * 1024 * 1024; // Size of the XCP server transmit queue in Bytes
 
 //-----------------------------------------------------------------------------
 // Calibration Segment

xcplib/src/a2l.c

@@ -27,6 +27,8 @@
 #include "xcp_cfg.h"   // for XCP_xxx
 #include "xcptl_cfg.h" // for XCPTL_xxx
 
+MUTEX gA2lMutex;
+
 static FILE *gA2lFile = NULL;
 
 static bool gA2lUseTCP = false;
@@ -1048,6 +1050,7 @@ bool A2lInit(const char *a2l_filename, const char *a2l_projectname, const uint8_
     memcpy(&gA2lOptionBindAddr, addr, 4);
     gA2lOptionPort = port;
     gA2lUseTCP = useTCP;
+    mutexInit(&gA2lMutex, false, 1000);
 
     // Check if A2L file already exists and rename it to 'name.old' if it does
     if (file_exists(a2l_filename)) {

xcplib/src/platform.c

@@ -404,7 +404,7 @@ bool socketStartup(void) {
     WSADATA wsaData;
 
     // @@@@ TODO: Workaround for Windows
-    mutexInit(&gWinMutex, true, 0);
+    mutexInit(&gWinMutex, false, 1000);
 
     // Init Winsock2
     wsaVersionRequested = MAKEWORD(2, 2);

xcplib/src/xcpEthServer.c

@@ -185,7 +185,7 @@ extern void *XcpServerTransmitThread(void *par)
     gXcpServer.TransmitThreadRunning = true;
     while (gXcpServer.TransmitThreadRunning) {
 
-        // Transmit all commmited messages from the transmit queue
+        // Transmit all committed messages from the transmit queue
         n = XcpTlHandleTransmitQueue();
         if (n < 0) {
             DBG_PRINT_ERROR("XcpTlHandleTransmitQueue failed!\n");

xcplib/src/xcpQueue32.c

@@ -54,7 +54,7 @@ Transport Layer segment, message, packet:
 #define CACHE_LINE_SIZE 64u // Cache line size, used to align the queue entries and the queue header
 
 typedef struct {
-    uint16_t dlc;     // lenght
+    uint16_t dlc;     // length
     uint16_t ctr;     // message counter
     uint8_t packet[]; // packet
 } tXcpMessage;
@@ -63,7 +63,7 @@ static_assert(sizeof(tXcpMessage) == XCPTL_TRANSPORT_LAYER_HEADER_SIZE, "tXcpMes
 
 typedef struct {
     uint32_t magic;                             // Magic number to identify the segment buffer
-    uint16_t uncommited;                        // Number of uncommited messages in this segment
+    uint16_t uncommitted;                       // Number of uncommitted messages in this segment
     uint16_t size;                              // Number of overall bytes in this segment
     uint8_t msg_buffer[XCPTL_MAX_SEGMENT_SIZE]; // Segment/UDP MTU - concatenated transport layer messages tXcpMessage
 } tXcpSegmentBuffer;
@@ -103,7 +103,7 @@ static void newSegmentBuffer(tQueue *queue) {
             i -= queue->queue_size;
         b = &queue->queue[i];
         b->size = 0;
-        b->uncommited = 0;
+        b->uncommitted = 0;
         queue->msg_ptr = b;
         queue->queue_len++;
         assert(queue->msg_ptr->magic == 0x12345678); // Check magic number
@@ -147,13 +147,13 @@ tQueueHandle QueueInit(uint32_t queue_buffer_size) {
     assert(queue->queue != NULL);
     for (uint32_t i = 0; i < queue->queue_size; i++) {
         queue->queue[i].magic = 0x12345678; // Magic number to identify the segment buffer
-        queue->queue[i].uncommited = 0;     // No uncommited messages
+        queue->queue[i].uncommitted = 0;    // No uncommitted messages
         queue->queue[i].size = 0;           // No data in this segment
     }
 
     DBG_PRINTF4("QueueInit: queue_buffer_size=%" PRIu32 ", queue_size=%" PRIu32 " (%" PRIu32 " Bytes)\n", queue->queue_buffer_size, queue->queue_size, queue->queue_buffer_size);
 
-    mutexInit(&queue->Mutex_Queue, 0, 1000);
+    mutexInit(&queue->Mutex_Queue, false, 1000);
 
     mutexLock(&queue->Mutex_Queue);
     queue->queue_rp = 0;
@@ -176,7 +176,6 @@ void QueueDeinit(tQueueHandle queueHandle) {
 
     clearQueue(queue); // Clear the queue
 
-    //_aligned_free(queue->queue);
     free(queue->queue);
     queue->queue = NULL;
     queue->queue_buffer_size = 0;
@@ -224,8 +223,8 @@ tQueueBuffer QueueAcquire(tQueueHandle queueHandle, uint16_t packet_size) {
         p->ctr = 0xEEEE; // Reserved value, indicates that this message is not yet commited
         p->dlc = (uint16_t)packet_size;
         b->size = (uint16_t)(b->size + msg_size);
-        b->uncommited++;
-        DBG_PRINTF5("QueueAcquire: size=%" PRIu16 ", uncommited=%" PRIu16 "\n", b->size, b->uncommited);
+        b->uncommitted++;
+        DBG_PRINTF5("QueueAcquire: size=%" PRIu16 ", uncommitted=%" PRIu16 "\n", b->size, b->uncommitted);
     } else {
         // No segment buffer available, queue overflow
         queue->packets_lost++;
@@ -253,11 +252,11 @@ void QueuePush(tQueueHandle queueHandle, tQueueBuffer *const queueBuffer, bool f
 
     tQueue *queue = (tQueue *)queueHandle;
 
-    DBG_PRINTF5("QueuePush: size=%" PRIu16 ", uncommited=%" PRIu16 "\n", queueBuffer->size, ((tXcpSegmentBuffer *)queueBuffer->handle)->uncommited);
+    DBG_PRINTF5("QueuePush: size=%" PRIu16 ", uncommitted=%" PRIu16 "\n", queueBuffer->size, ((tXcpSegmentBuffer *)queueBuffer->handle)->uncommitted);
 
     mutexLock(&queue->Mutex_Queue);
 
-    ((tXcpSegmentBuffer *)queueBuffer->handle)->uncommited--;
+    ((tXcpSegmentBuffer *)queueBuffer->handle)->uncommitted--;
 
     tXcpMessage *p = (tXcpMessage *)(queueBuffer->buffer - XCPTL_TRANSPORT_LAYER_HEADER_SIZE);
     assert(p->dlc > 0 && p->dlc <= XCPTL_MAX_DTO_SIZE);
@@ -317,7 +316,7 @@ tQueueBuffer QueuePeek(tQueueHandle queueHandle, bool flush, uint32_t *packets_l
         }
 
         // Return tail segment buffer if it is not empty, fully committed and there are more segments in the queue
-        if (!(queue->queue_len > 1 && b->uncommited == 0 && b->size > 0)) {
+        if (!(queue->queue_len > 1 && b->uncommitted == 0 && b->size > 0)) {
             b = NULL;
         }
     }
@@ -366,7 +365,7 @@ void QueueRelease(tQueueHandle queueHandle, tQueueBuffer *const queueBuffer) {
 
     DBG_PRINTF5("QueueRelease: size=%" PRIu16 "\n", queueBuffer->size);
 
-    // Free this segment buffer when succesfully sent
+    // Free this segment buffer when successfully sent
     mutexLock(&queue->Mutex_Queue);
     if (++queue->queue_rp >= queue->queue_size)
         queue->queue_rp = 0;