FIFO Thread Safe Variables

[edinsofty1]

Hi,

I am currently trying to develop a Datalogger Application for a Fibre Optic link running at 250Mb/s. I have the code sorted that logs the data, but now have to try and store the data to disk.

I need 3 Rx threads to keep up with the incoming data to ensure I dont get datalink errors on the card buffer, and I store the data into a Link List for processing by a Write Thread. To do this, I am trying to create a Thread Safe linked list variable as below :

typedef struct list List;

struct list {
 int  length;
 Node  *head;
 Node *current;
};

DefineThreadSafeVar(List, SafeList);  

List *SafeListPtr;  

Then in the main function :

InitializeSafeList();

I then have an initialise Linked List function as follows:

ViUInt32 LL_Initialise(void)
{
 SafeListPtr = GetPointerToSafeList();
 
 SafeListPtr = (List*)malloc(sizeof(List));

 SafeListPtr->length = 0;
 SafeListPtr->head = SafeListPtr->current = NULL;
 
 ReleasePointerToSafeList();
 
 return 0;
}

This appears to initialise the linked list correctly, setting the head and current pointers to NULL.

But, when i call the following function :

ViUInt32 LL_GetLength()
{
 ViUInt32 length;
 
 SafeListPtr = GetPointerToSafeList();     <- ******************

 length = SafeListPtr->length;
 
 ReleasePointerToSafeList();
 
 return (length);
}

They line I have marked causes the head and current pointers to return to Uninitialised and then causes a Run Time error when try to use them.

If you have followed that.....Any ideas?

Thanks

Stuart

[RobertoBozzolo]

I lost myself a bit into your TSV structure, but more conceptually I suppose you want to write data to disk as they are acquired by your acquisition threads: can't this task be better approached using a Thread Safe Queue? TSQ mechanism can take care for you of the FIFO order you seem to look for.

Also, take a look at this example that may provide you a valid alternative, together with an example of using TSQs.

[edinsofty1]

Thanks. I was having a look at the TSQ as you replied! As I have 3 threads receiving data, the queue will be filling with the data in a possible random order. The FibreXtreme data card I am using passes a Sequence Number and a Start Byte Number with each packet it receives to allow for multi threading logging.How easy would it be to sort the data in the queue into the correct order before writing to disk bearing in mind the data has been getting written from 3 seperate threads? Cheers Stuart

[carlox]

Stuart, the variable SafeList is created and allocated when you call 

InitializeSafeList();

So there is no need to allocate it and assign to SafeListPtr.

The correct code for the function is as follows

ViUInt32 LL_Initialise(void)
{
 SafeListPtr = GetPointerToSafeList();
 SafeListPtr->length = 0;
 SafeListPtr->head = SafeListPtr->current = NULL;
 
 ReleasePointerToSafeList();
 
 return 0;
}

[RobertoBozzolo]

Well, I suppose you can create an intermediate buffer in the receiving function:

• create an array of your List structure
• read a record from the queue: if in the correct order, stream it to disk
• if out of order, fill a record of the array
• when you receive the next record in the correct order:
• put it to disk
• sort array content (e.g. with QuickSort passing an appropriate comparison function) and see if and how many records in the buffer can be put to disk

You may want to design a different algorithm to decide when to sort so that you don't have too much data in the intermediate buffer: sorting on every record may cause too much overhead in your application and a growing buffer of record left to write to disk.

Please keep posting: this multi-producer-one-consumer framework is an interesting schema to study! 😉

[edinsofty1]

Hi,

I managed to uses a TSQ to carry out the FIFO operation, and as each DMA transfer from the FibreXtreme card comes with a Sequence Number & Byte Number, I should be able to write to a specific area of the file ( I am trying to use fseek to achieve this but not sure if it's working that well - any better method?).

Another issue with this process is with "fwrite". The program runs happily for a few seconds with 3 Rx threads and 3 "write to disk" threads. Then, the fwrite command slows right down and the FIFO queue fills up and runs out of memory. If I remove the fwrite command, the threads run happily, so I can only assume the fwrite is causing this. My concern is why it writes in the region of 200Mbytes then starts to slow down?

	while (!m_quitPressed)
	{
		CmtGetTSQAttribute (queue2Handle, ATTR_TSQ_ITEMS_IN_QUEUE, &length2); 

		if (length2 > 0)
		{
			// Read a packet from the Linked List			
			numRead = CmtReadTSQData (queue2Handle, rxPacket, length2, 0, 0);

			// Write data to file - fileHandle         
			elemsWritten = fwrite(rxPacket, 4, (BUFFER_SIZE/4), fileHandle2);   
		}
		
	}

I have tried all variations of BUFFER_SIZE, but made no difference.

Any ideas?

[RobertoBozzolo]

I have no precise answer for you last question: 200 MB should not be a problem with modern systems but if you are low in memory it could be difficult for the system to sustain that update rate. Periodically flushing the file with fflush () could make the system more stable: you will need to properly dimension the queue since flushing the file will take some time so the buffer must be properly prepared. Alternatively, could you segment the process in chunks that produce smaller files?

[edinsofty1]

After some advice, I am looking at trying to implement the datalogger a slightly different way, but am having issues with the TSQ.

I now only have 1 TSQ but this needs to be accessed by 3 threads. I thought because it was thread safe, CVI would handle who could write to the TSQ, but it throws up a run-time error saying "this or another thread is currently writing to the TSQ". How can I get round this? Thread locking or anything more efficient?

[RobertoBozzolo]

This error is expected given the design principle of TSQs: see this page for reference.

Thread locks have been specifically designed to prevent concurrent access to shared resources in multithreadeed applications: it seems to me that this is the preferred way of addressing this problem. Please note that you should not use named locks or process events while waiting for a lock if you want the maximum efficiency.