I would like to create a buffer into which a thread can push incoming data and the main thread can extract it for processing. Data are coming from the serial port in a worker thread and should be consumed by the main thread. The data is a byte stream.

Are there any good examples available?
I have only implemented such systems in C on embedded platforms so it is not pascal enough to easily port.

Packager LazUtils
unit LazCollectionss

TLazThreadedQueue

So what you meant is that it is part of Lazarus already? I found it here in my local installation:

~/dev/lazarus/2.0.8/components/lazutils/lazcollections.pas

When I added lazcollections to the uses clause of my test project it loaded fine when right-clicking and selecting "Find Declaration.." So it is clearly in the search path for Lazarus. :-)

Now to my question:

Is there some *example* around for using TLazThreadedQueue as a circular buffer?
In the examples dir are only examples for LazUnicode and LookUpStringList...
And when reading the sources I cannot really say I understand what it does or how it works.

Is it even applicable for storing data such as a sequence of bytes arriving via a serial port?
The names PushItem and PopItem really suggest it works like a LIFO buffer rather than a FIFO buffer, which is what I need...

LIFO: Last In First Out, like how a stack works
FIFO: First In First Out, like a tube where the items reside for a while but come out in orderly sequence.

It is FIFO. Its a queue, not a stack.

Look at my latest commits in Trunk to fpdebug > FpDbgUtil I use the queue there.

Note:
- Calling "grow" when the buffer is NOT empty, works in latest trunk only. Actually you want that fix anyway, because on Win there is a race condition, that could starve individual threads.
r 63811 and 63833 (must take both / I introduced a silly type in the first)

- Calling shutdown (to stop all threads from waiting), may also not work, if there are many threads (at least on win). But the fix is in trunk, and easy to port. r 63860

You need a type (e.g. class, or array, or pointer) that you can then push. So if you get bytes from an input, I suggest you push collected chunks of them (fixed or dynamic size).

A "simplest case" usage example would be nice to have. I still when looking at the sources have no clue as to how it would be used...

It seems to me like at the very least one should define what type of data will be stored in the buffer...
Is this how you use it?

var
  MyFifoBuffer: TLazThreadedQueue;  77Whgat do I do about the <T>???
  b: byte;
  i: integer
begin
  MyFifoBuffer := TLazThreadedQueue.Create($8000); //Why no type???
  for i:= 0 to 255 do
  begin
    b := i and 255;
    MyFifoBuffer.PushItem(b);
  end;
  for i := 0 to 255 do
  begin
    b := MyFifoBuffer.PopItem;
    Writeln(IntToHex(b,2);
  end;
end.

And how do you use ShutDown, DoShutDown, QueSize and Grow?

Very confused.

Bo Berglund wrote

A "simplest case" usage example would be nice to have.

True. Once you got enough info from this thread, please write one.

Ok, I see. It's a generic, so this question is not directly related to the TLazThreadedQueue

type
   TMyData = class
       FData: array of byte;
    end;
   TMyThreadedQueue = specialize TLazThreadedQueue<TMyData>;

var
  a: TMyData
  q: TMyThreadedQueue

q := TMyThreadedQueue.create($8000);
a := TMyData.Create;
q.PushItem(a);

Of course you can do
  TMyData = array of byte;  // or record .... end if you want   TMyThreadedQueue = specialize TLazThreadedQueue<TMyData>;

or even
   TMyThreadedQueue = specialize TLazThreadedQueue<Byte>;

Pushing arrays (dyn ar stat) is fine.

Pushing individual bytes => works. And if you get a byte once or twice in a second then it is ok.

If you get bytes every few milliseconds or faster, then pushing bytes may be too expensive (I don't know, but I would expect at some point....). In order to push (or pop) the queue has to obtain a lock. You have to make sure your computer can handle that fast enough. (maybe it can....)

And how do you use ShutDown, DoShutDown, QueSize and Grow? Very confused.

Grow gives a positive/negative delta. It is added to the current size. But as I said, for this to work you need the patches. Also "grow" on a live queue, requires to get the lock yourself. This is not automated. (Might be an idea to add that....)

- If the queue is full then PushItem will wait until timeout, or until something is popped.
- if the queue is empty PopItem will wait in that manner. If threads are waiting, and ShutDown is called, then Push/PopItem return wrAbondoned. (So you need to check for this)

Keep in mind, that if you push an array (or anything containing an array), you push a reference.

That is any modification you make to the array content after you pushed, will modify the date in the queue. (reference to the same data)

To prevent that you must ensure the data has only one reference holder. See SetLength
https://www.freepascal.org/docs-html/rtl/system/setlength.html

Bo,

Most of the users on this forum have never interfaced with hardware , as you can see from the responses. Those who have interfaced with hardware have developed circular buffers which they know work , and they reuse them again and again.

I will describe in general terms the structure a circular buffer implemented to receive data , as it is usually more difficult to received the data than to send it due to the asynchronism between the sender and receiver. It has been used for many years , starting in 16bit-DOS , the 32bit-DPMI using interrupt handlers , then Linux using threads. The concept is the same for versions.

The following assumes you are receiving data from a sender , and in this case the sender and receiver are connected via 1Gbps Ethernet using Synapse (works really well).

A thread receives data into an array of byte. RxBuffer : Array[0..1500] of byte ... and copies the number of bytes received into a circular buffer , which is an array of RxBuffer.

In pseudo code it looks like this ...
CircularBuffer : Array[0..N] of RxBuffer
There is a ReadIndex and a WriteIndex , both longword or longint.

The thread moves the data into the circular buffer continuously  increase
the WriteIndex each write of RxBuffer data. If the WriteIndex > N then
WriteIndex := 0 , otherwise it is increased by 1.

The main loop (below) moves the byte data using MOVE from CircBuffer to whatever structure it represents , increase the ReadIndex by one and decodes the data as needed.

 If ReadIndex  WriteIndex then
  .. do the work as described above and increment the ReadIndex

I use the Free Pascal unit which allows suspending the thread while the ReadIndex is being increased. In the old DOS/DPMI days we would disable interrupts briefly.

If you are innterested I can send you code snippets showing exactly how to implement the circular buffer.

Regards
Brian

I did this about 10 years ago when working on a PIC project with serial data coming in from sensors and such, using Ansi-C. And the buffer was just an array of char (byte in pascal) and there was a read and a write index. The read index was only ever changed by the consumer and the write index by the interrupt routine when adding data.

The data reception was done in the interrupt and the code stuffed the data into the array using ++ syntax for the index, with a check for overflowing the array and then setting the index back to zero.

There was probably also an overrun detection and action if the consumer is not fast enough, but I do not remember now...

This is where I lose you, are you saying the circular buffer is not individual bytes but something larger? What is the size of RxBuffer? Looks like it is an array of byte, can you then make an array of such arrays?

Looks like an array of some bigger datatype then. But if so then one needs to be able to detect in the receiver that enough data has arrived to make one such object to put onto the next buffer...

Do you use CriticalSection for this?

If you are innterested I can send you code snippets showing exactly how to implement the circular buffer.

Please do, thanks!

Bo, Most of the users on this forum have never interfaced with hardware , as

you

can see from the responses. Those who have interfaced with hardware have developed circular buffers which they know work , and they reuse them

again

and again.

I did this about 10 years ago when working on a PIC project with serial data coming in from sensors and such, using Ansi-C. And the buffer was just an array of char (byte in pascal) and there was a read and a write index. The read index was only ever changed by the consumer and the write index by the interrupt routine when adding data.

The data reception was done in the interrupt and the code stuffed the data into the array using ++ syntax for the index, with a check for overflowing the array and then setting the index back to zero.

There was probably also an overrun detection and action if the consumer is not fast enough, but I do not remember now...

This is where I lose you, are you saying the circular buffer is not individual bytes but something larger? What is the size of RxBuffer? Looks like it is an array of byte, can you then make an array of such arrays?

... The CicBuffer is an array of N RxBuffer , which is an array of byte. The main loop extracts the byte array and either decodes them or type casts them into a record structure if it is compatible with the contents of the byte array,

Type
 RxBufferType = array[0..1500] of byte;
Var
 CirBuffer : array[0..N] of RxBufferType;

decodes

the data as needed.

Looks like an array of some bigger datatype then. But if so then one needs to be able to detect in the receiver that enough data has arrived to make one such object to put onto the next buffer...

... you appear to be thinking of a asynchronous serial (RS-232 / 422) with no delimiters to indicate when a complete message is received.

... think of the byte array as an Ethernet packet of any length data of which the data structure is known. This is received using Synapse (UDP in this case) and copied into RxBuffer byte array by a thread which is always running until the main program is shut down. Whenever the thread receives a "complete" packet from Synapse (it signals a complete packket) , the thread code then puts it into RxBuffer and then copies RxBuffer into CircBuffer[WriteIndex] and increments the WriteIndex. If the WriteIndex < N then WriteIndex := WriteIndex +1 else WriteIndex := 0 // wraps

The ReadIndex is incremented similar to the WriteIndex, if ReadIndex < N
then ReadIndex := ReadIndex +1 else ReadIndex := 0;

The circular buffer can be made as large as you need to "absorb" incoming data while the main loop is busy doing something else. The "nice" aspect of a circular buffer is that if the main loop is very slow and the incoming data stream is very fast , the WriteIndex will eventually overrun the ReadIndex and data is lost but nothing chokes or crashes , and eventually the ReadIndex will catch up.

BTW : There are enough things that can go wrong implementing a circular buffer , and while it should be possible (and very nice) to implement it as an object , it may not be worth the effort. At least get it working as non-object code , then decide.

Do you use CriticalSection for this?
No . .. I started to originally but found it wasn't necessary as it uses a semaphore to prevent a Write occurring when the ReadIndex is being incremented. The FPC unit syncobjs is used the create a semaphore which is RESET before incrementing the ReadIndex and SET immediately after. This prevents a potential race condition where data could be written to CircBuffer at the exact time the ReadIndex is being incremented.

If you are innterested I can send you code snippets showing exactly how to implement the circular buffer.

BTW : The circular buffer approach can also be used when transmitting data. For example Synapse maintains its own transmit buffers , which signal when you can send more data.

<quote author='Free Pascal - General mailing list'> I would like to create a buffer into which a thread can push incoming data and the main thread can extract it for processing. Data are coming from the serial port in a worker thread and should be consumed by the main thread. The data is a byte stream.

Are there any good examples available?
I have only implemented such systems in C on embedded platforms so it is not pascal enough to easily port.