FIFO buffer, concurrent tasks

[denis90]

Hi there,
my problem here is more a logical one than an implementation (which will be done in cpp) one, hence this could look a bit out of topic here, I apologize, but I don't know any other good programming related forums.

I'd like to implement a FIFO buffer for an application of mines, but since this buffer is going to be used by more that one simultaneous task, I got some trouble. I'll try to explain the situation in the text below as clear as I can.

I am dealing with a multitasking environment, the tasks involved are basically these:

1) main()
2) interrupt()

main() task basically poll the buffer, whether it is not empty, it removes whatever is in it by calling FifoBuffer::pop().

interrupt() task is something really "hot", main() doesn't know whether interrupt() is being called (is an hardware interrupt actually), and its execution can't be stopped for any reason. This can, by the way, stop main() from its execution. This function basically use the buffer by pushing something in it, that is calling FifoBuffer::push().

So, main() just run over and over, being spuriously interrupted by interrpt(). As interrupt() finishes its execution, the control is given back to main().


A simple FIFO buffer made up of an array and 2 pointers (like first and last) won't work for this situation since this could lead to some serious synchronization problems when the buffer gets either empty or full.
The problem here is that it is possible for FifoBuffer::pop() to get stopped in the middle of its execution by an interrupt that will then call FifoBuffer::push(). When it's done, control would be given back to FifoBuffer::pop().
Also it is not possible to lock the buffer since when an interrupt occurs, FifoBuffer::push() has to finish its job as fast as it can, and really can't neither wait for FifoBuffer::pop() to get done or give back the control to FifoBuffer::pop() to finish its job.

My first idea here was to use dual buffering, so that I could lock the buffer being used by FifoBuffer::pop(), then if an interrupt occurs while pop() is being executed, FifoBuffer::push() could use the other buffer, which is ought to be free. Anyway, this way I would loose track of the pushed objects, with the buffer becoming a random buffer instead of a FIFO. Someway to solve this problem would be another small FIFO buffer to keep track of pushes, but this means that another FIFO is to be implemented, and the problem would be the same I started from.

I really can't figure this out, it's blowing my mind, please help. Thank you.

[PanGalactic]

Without locks, you have to use atomic operations for this to work. You can look at how the Linux kernel deals with interrupts...

[Disch]

Don't count on code being atomic because there's not really a way to guarantee it. Use a locking mechanism.

Are you using a multithreading library or anything? If so, this problem is easily solved with Mutexes.

Just associate a mutex with your FIFO buffer, and lock it before each access (and unlock it afterwards).

[PanGalactic]

@Disch: In an OS kernel, you generally cannot sleep/block in an interrupt handler. Locks are therefore not allowed there. That is certainly the case with the Linux kernel.

The only peculiarity is that a handler runs at interrupt time and therefore suffers some restrictions on what it can do. ... Handlers also cannot do anything that would sleep, such as calling sleep_on, allocating memory with anything other than GFP_ATOMIC, or locking a semaphore. Finally, handlers cannot call schedule.

http://www.xml.com/ldd/chapter/book/ch09.html

[Disch]

Ouch.

Maybe don't do this in an interrupt then? Maybe just do it in another thread?

volatile unsigned interruptspending = 0;

void interrupt()
{
   ++interruptspending;
}

void SomeOtherThread()
{
   while(program_running)
   {
      while(interruptspending > 0)
      {
         --interruptspending
         LockMutex();
         ManipulateFIFO();
         UnlockMutex();
      }

      YieldToMainThread();
   }
}

void MainThread()
{
  // go about your business
}

Seems sort of hackish... but it could work.

[imi]

So you got the following requirements, right?

1) "interrupt" can interrupt "main" unconditionally at any non-atomic instruction.
2) No other interruptions: "main" can not interrupt "interrupt". "interrupt" can not interrupt itself. "main" can not interrupt itself.
3) "interrupt" always call "push", "main" always call "pop"
4) "interrupt" must not do any fancy things that would block, including allocating memory on the heap (since new usually uses malloc and malloc usually do some locking stuff).

As of requirement 4), it is not possible to find a solution that works perfectly in the case of buffer overflows ("interrupt" could be executed until any pre-allocated buffer overflows)

A simple FIFO buffer made up of an array and 2 pointers (like first and last) won't work for this situation since this could lead to some serious synchronization problems when the buffer gets either empty or full.
The problem here is that it is possible for FifoBuffer::pop() to get stopped in the middle of its execution by an interrupt that will then call FifoBuffer::push(). When it's done, control would be given back to FifoBuffer::pop().

I fail to see a problem with empty buffers.

If pop and push can run concurently, everthing is fine. Actually, it should be even less problematic that other multi-threading scenarios because of 2) and 3)

Use a ring-buffer and let "push" only writes to the write-pointer whereas "pop" only writes the read-pointer.

So if you can live with the problem of buffer overflows, something along these lines would do the trick. I assume you pass int's along and "-1" is returned if the stack is empty.

struct Fifo
{
    int buffer[100];
    volatile int read, write;
    Fifo() : read(0), write(0) {}

    void push(int v)
    {
        if (write+1 == read) abort(); // damnit - buffer overflow.
        buffer[write++] = v;
        if (write >= 100) write = 0;
    }

    int pop()
    {
        if (read == write) return -1; // stack is empty
        int r = buffer[read++];
        if (read >= 100) read = 0;
        return r;
    }
};

(Beware: I haven't tested the code)

Anyway, I am only assuming that reading out the value of "write" is atomic, which really should be true.


If you really can't live with the fear of a buffer overflow, you can reduce the risk by allocating a new buffer together with new pointers in main() (of course, that means in "pop") once the buffer is near-full and then swap to the new buffer+pointers in one atomic statement. You can do this by encapsulating all three values in an struct. You have to keep the old buffer and use this until it is empty (as you don't know how many items interrupt will insert while you create and initialize the new buffer...)

struct Buffer { vector<int> b; volatile int read, write; Buffer(int s):b(s),read(0),write(0){} };
struct Fifo
{
    int size; // current buffer size
    Buffer* pBuffer;
    Buffer* pOldBuffer; // used during switch to new buffer

    ... // like above, but allocate pBuffer in ctor and use pBuffer->*. destructor frees the buffers

    int pop()
    {
        if (!pOldBuffer && pBuffer is nearly full)
        {
            size *= 2;
            pOldBuffer = pBuffer;
            pBuffer = new Buffer(size); // one atomic switch to new and empty buffer!
        }

        Buffer* b = pOldBuffer ? pOldBuffer : pBuffer;
        // from now on, use b->* as in pop above
        // Don't forget to free & reset pOldBuffer before return, should it be empty now.
    }
};

The code is completely untested and written off the cuff. Excpect some typos :-D. But I hope you get the basic idea.

Ciao, Imi.

[denis90]

@ PanGalactic : I'll try to take a look at that, even if it's some very complicated and long code (I suppose) I'll try to, thanks for pointing that out!


@ Disch : Unfortunately I have to do it in an interrupt handling routine, the program is actually running on a bare-metal processor, there is no operating system at all. Thanks for your reply


@ imi :

So you got the following requirements, right?

1) "interrupt" can interrupt "main" unconditionally at any non-atomic instruction.
2) No other interruptions: "main" can not interrupt "interrupt". "interrupt" can not interrupt itself. "main" can not interrupt itself.
3) "interrupt" always call "push", "main" always call "pop"
4) "interrupt" must not do any fancy things that would block, including allocating memory on the heap (since new usually uses malloc and malloc usually do some locking stuff).

As of requirement 4), it is not possible to find a solution that works perfectly in the case of buffer overflows ("interrupt" could be executed until any pre-allocated buffer overflows)

You got the idea, that's right. Anyway I really don't care about buffer overflows, should it ever occur, I will enlarge the buffer allocation statically.

I fail to see a problem with empty buffers.


If pop and push can run concurently, everthing is fine. Actually, it should be even less problematic that other multi-threading scenarios because of 2) and 3)

Use a ring-buffer and let "push" only writes to the write-pointer whereas "pop" only writes the read-pointer.

I fail too while looking at your code, that's a nice, short, simple and clean piece of code. Some days ago I managed to write down a more complicated and stupid implementation of the same buffer, which uses instead some in-class flags like is_empty and is_full. This really accomplished the law and introduced sync problems that would have been occured whether the buffer would had been empty.

The code is completely untested and written off the cuff. Excpect some typos :-D. But I hope you get the basic idea.

Yea, I got it! Thank you, you saved my day :)

[imi]

I looked on the code again and immediately spotted a problem in write+1 == read. Shame on me. Of course, it should be (write+1)%100 == read ;-)

Ciao, Imi.