Take the 2-minute tour ×
Programmers Stack Exchange is a question and answer site for professional programmers interested in conceptual questions about software development. It's 100% free, no registration required.

I had a job process that was executing a lot of IO to read and write temporary files.
Now I want to (need to) reduce the quantity of IO executions.

So I want to create a sort of circular buffer that is going to be filled up with a data from a text file within first thread.

The consumer (reading) thread will fetch data from this buffer.

The problem is that there could be multiple consumers that need to read from same one buffer.

For buffer implementation I prefer not to use any existing wrappers (simple array, where i just "play" with indices is enough).

I also don't want to create a separate buffer for every reader. And ofcourse I want to avoid any deadlocks and unnecessary blockings.

UPDATE
Right now I am using circular buffer (an array and 2 indices)

The question is how to implement such buffer that can be accessed by multiple consumers where each consumer can read from it interdependently from other consumers (one costumer may read faster than the other one).

IMPORTANT UPDATE
The first thread don't know (and should not know) about it's consumers!!! It should write data to a buffer, when data ends it should raise a flag.

share|improve this question
    
Please expand what your question is here. Are you not sure how to implement such a buffer? What part is unclear to you? What experience do you have? Do you know about the producer/consumer problem? Monitors? Have you done anything with the Runnable interface and synchronized methods? –  sebastiangeiger Dec 4 '11 at 13:06
    
Take a look at code.google.com/p/disruptor which implements allows multiple threads to access input data-streams based on a circular buffer. –  Patrick Dec 4 '11 at 13:58
    
@Patrick, thanks but I asked for solution without any "wrappers" (this includes 3rd party libs) –  kenny Dec 4 '11 at 14:10
    
I wasn't suggesting that you should use it. Read the documentation and the code. –  Patrick Dec 4 '11 at 16:14
    
@kenny As far as I can tell you are filling the buffer once and then just removing elements. Why do you need a ring buffer for that? –  sebastiangeiger Dec 4 '11 at 16:34

5 Answers 5

Using a bare array is forcing you to do more work than you need to. People have spent a lot of time coming up with data structures to make your life easier, so use them.

It sounds like you are trying to implement a Queue. This is a data structure that allows you to add things to the collection and then remove them later. This is done in a first in first out order. Java even has a Thread-Safe Queue built in. You can simply make an instance of that class and give each of your threads a reference to it. The producer will add items to the queue and the consumer(s) will remove items from the queue or perform another action if the queue is empty.

If you really have your heart set on implementing everything yourself, you will need to synchronize all access to your array so that two threads can't access the buffer at the same time.

EDIT:

It wasn't clear before that you wanted every consumer to read every bit of data. You can extend the Queue implementation so it keeps track of the count of items (making size() constant time) and then throw an exception if the queue is full.

To solve the popping issue is a little trickier. You need to keep track of the index that each consumer is at, a map would do this. Then when all of the consumers have moved passed an item, it can be popped and update all the indexes in the map. When a consumer calls pop() and the item needs to be retained you can re-route it to a peek().

share|improve this answer
    
but again it doesn't solve the main problem: the buffer must have a limit, let's say size of N items. if i have circular buffer with 2 pointers and one consumer it's not a problem to "simulate" FIFO. but when there are X consumers how do I know when I should "popout" item that was already read by all consumers? –  kenny Dec 4 '11 at 14:42
    
@kenny: Made an edit based on your comment. –  unholysampler Dec 4 '11 at 15:31

On the Java side look at the disruptor which is a non-blocking, high Performance Inter-Thread Messaging Library. This is in contrast to the 'traditional' lock based concurrency that most folks still use with Java/JVM programming. The style that you want to avoid in your use case (why deal with shared state when you don't have to).

share|improve this answer
    
I assume you meant ring buffer? –  Phonon Apr 26 '13 at 0:35
    
Errr yeah PEBCAK –  Martijn Verburg Apr 26 '13 at 9:25

I have created such a class as you describe in C++, but not Java. I used a singleton pattern, combined with templates so that if you knew the buffer type and size you will get the correct instance. So you can have different buffers of different types at the same time. I'm not sure you'll be able to do this exactly in Java or not. But regardless of this, what I say below still holds.

Implementation wise you'll need some semaphores and mutexes. I used two semaphores, one to count the number of items in the buffer and one to count the amount of space left. You can then use these in your put() and get() methods to determine if there is space left or anything to take. One you've done that you can grab the mutex that protects access to the buffer, copy out your object and change your index pointers.

Using the semaphores avoids deadlocking on the mutex in the scenario where the getter thread acquires it before the putter thread has placed anything in the buffer. You can use the index pointers to implement a numItems() function and you'll need them to workout the array indies to use for the get and put.

share|improve this answer
    
I still can't understand how do I know it the item inside the buffer was been read by all consumers and it can be removed? –  kenny Dec 4 '11 at 14:57
    
@kenny, Sorry I must have misunderstood, I assumed that you had multiple consumers but that they each didn't need every item in the buffer, just what was there on the head of the queue at the time. I think this is a much more difficult publish-subscribe problem. –  fwgx Dec 4 '11 at 15:00
    
yes, multiple number of consumers (number of consumers is determined on runtime), read same data from the producer. (each consumer will do a different job on this input data). The producers is not aware of how many consumers it has, it can only signal that it's done writing data to buffer. –  kenny Dec 4 '11 at 15:06
    
We do this at work using RTI's version of Data Distribution Service (DDS). There is an open source version too: prismtech.com/opensplice/products/community-edition-open-source but this starts getting reasonably complicated. I've never used the OpenSplice version but I know I've spent a long time fine tuning the QoS and creating simplified wrappers for the RTI implementation. It does scale pretty well and allows producer-consumer operation over the network using multicast. –  fwgx Dec 4 '11 at 15:18

The question is how to implement such buffer that can be accessed by multiple costumers where each costumer can read from it interdependently from other costumers (one costumer may read faster than the other one).

In order to do this safely, you'll need to ensure that only one thread can access the buffer at a given time. This can be a complex problem, but Java's synchronized methods simplifies things for you. In a nutshell, if one thread is executing a synchronized method of an object, no other synchronized method on that object may be executed in any thread until that method completes.

So, your job is to create a class for your buffer with synchronized read and write methods. Instantiate that class early in your program and share the resulting object with all producers and consumers. Java's method synchronization will ensure that only one thread accesses the object at a time.

share|improve this answer

what unholysampler says + you MUST know all consumers (because without it you don't know whether all of them actualy read some byte), and if there is one that reads slower than others (and than producer produces), you will end with full array not read by him.

You can either 1) on producer side: throw or block 2) remove him as consumer and start his own queue (but use something that can grow) 3) remove him as consumer and throw exception when he wants to read.

share|improve this answer

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

Not the answer you're looking for? Browse other questions tagged or ask your own question.