haiku.git

/******************************************************************************

	File:	BufferStream.h

	Copyright 1995-97, Be Incorporated

******************************************************************************/
#ifndef _BUFFER_STREAM_H
#define _BUFFER_STREAM_H

#include <stdlib.h>
#include <OS.h>
#include <SupportDefs.h>
#include <Locker.h>
#include <Messenger.h>


class BSubscriber;


/* ================
   Per-subscriber information.
   ================ */

struct _sbuf_info;

typedef struct _sub_info {
  _sub_info			*fNext;		/* next subscriber in the stream*/
  _sub_info			*fPrev;		/* previous subscriber in the stream */
  _sbuf_info		*fRel;		/* next buf to be released */
  _sbuf_info		*fAcq;		/* next buf to be acquired */
  sem_id			fSem;		/* semaphore used for blocking */
  bigtime_t			fTotalTime;	/* accumulated time between acq/rel */
  int32				fHeld;		/* # of buffers acq'd but not yet rel'd */
  sem_id			fBlockedOn;	/* the semaphore being waited on */
								/* or B_BAD_SEM_ID if not blocked */
} *subscriber_id;


/* ================
   Per-buffer information
   ================ */

typedef struct _sbuf_info {
  _sbuf_info		*fNext;		/* next "newer" buffer in the chain */
  subscriber_id		fAvailTo;	/* next subscriber to acquire this buffer */
  subscriber_id		fHeldBy;	/* subscriber that's acquired this buffer */
  bigtime_t			fAcqTime;	/* time at which this buffer was acquired */
  area_id			fAreaID;	/* for system memory allocation calls */
  char				*fAddress;
  int32				fSize;     /* usable portion can be smaller than ... */
  int32				fAreaSize; /* ... the size of the area. */
  bool				fIsFinal;	/* TRUE => stream is stopping */
} *buffer_id;


/* ================
   Interface definition for BBufferStream class
   ================ */

/* We've chosen B_MAX_SUBSCRIBER_COUNT and B_MAX_BUFFER_COUNT to be small
 * enough so that a BBufferStream structure fits in one 4096 byte page.
 */
#define	B_MAX_SUBSCRIBER_COUNT	52
#define	B_MAX_BUFFER_COUNT		32

class BBufferStream;
class BBufferStreamManager;

typedef BBufferStream* stream_id;  // for now


class BAbstractBufferStream
{
public:
#if __GNUC__ > 3
	virtual				~BAbstractBufferStream();
#endif

	virtual	status_t	GetStreamParameters(size_t *bufferSize,
											int32 *bufferCount,
											bool *isRunning,
											int32 *subscriberCount) const;

	virtual	status_t	SetStreamBuffers(size_t bufferSize, 
										 int32 bufferCount);

	virtual	status_t	StartStreaming();
	virtual	status_t	StopStreaming();

protected:

virtual	void		_ReservedAbstractBufferStream1();
virtual	void		_ReservedAbstractBufferStream2();
virtual	void		_ReservedAbstractBufferStream3();
virtual	void		_ReservedAbstractBufferStream4();

	friend class BSubscriber;
	friend class BBufferStreamManager;

	virtual stream_id	StreamID() const; 
	/* stream identifier for direct access */

	/* Create or delete a subscriber id for subsequent operations */
	virtual	status_t	Subscribe(char *name,
								  subscriber_id *subID,
								  sem_id semID);
	virtual	status_t	Unsubscribe(subscriber_id subID);

/* Enter into or quit the stream */
	virtual	status_t	EnterStream(subscriber_id subID, 
									subscriber_id neighbor,
									bool before);

	virtual	status_t	ExitStream(subscriber_id subID);

	virtual BMessenger*	Server() const;	/* message pipe to server */
	status_t 		SendRPC(BMessage* msg, BMessage* reply = NULL) const;
};


class BBufferStream : public BAbstractBufferStream
{
public:

						BBufferStream(size_t headerSize,
									  BBufferStreamManager* controller,
									  BSubscriber* headFeeder,
									  BSubscriber* tailFeeder);
		virtual			~BBufferStream();
				
/* BBufferStreams are allocated on shared memory pages */
		void			*operator new(size_t size);
		void			operator delete(void *stream, size_t size);

/* Return header size */
		size_t			HeaderSize() const;

/* These four functions are delegated to the stream controller */
		status_t		GetStreamParameters(size_t *bufferSize,
											int32 *bufferCount,
											bool *isRunning,
											int32 *subscriberCount) const;

		status_t		SetStreamBuffers(size_t bufferSize, 
										 int32 bufferCount);

		status_t		StartStreaming();
		status_t		StopStreaming();

/* Get the controller for delegation */
		BBufferStreamManager *StreamManager() const;

/* number of buffers in stream */
		int32			CountBuffers() const;

/* Create or delete a subscriber id for subsequent operations */
		status_t		Subscribe(char *name, 
								  subscriber_id *subID,
								  sem_id semID);

		status_t		Unsubscribe(subscriber_id subID);

/* Enter into or quit the stream */
		status_t		EnterStream(subscriber_id subID, 
									subscriber_id neighbor,
									bool before);

		status_t		ExitStream(subscriber_id subID);

/* queries about a subscriber */
		bool			IsSubscribed(subscriber_id subID);
		bool			IsEntered(subscriber_id subID);

		status_t		SubscriberInfo(subscriber_id subID,
									   char** name,
									   stream_id* streamID,
									   int32* position);

/* Force an error return of a subscriber if it's blocked */
		status_t		UnblockSubscriber(subscriber_id subID);

/* Acquire and release a buffer */
		status_t		AcquireBuffer(subscriber_id subID, 
									  buffer_id *bufID,
									  bigtime_t timeout);
		status_t		ReleaseBuffer(subscriber_id subID);

/* Get the attributes of a particular buffer */
		size_t			BufferSize(buffer_id bufID) const;
		char			*BufferData(buffer_id bufID) const;
		bool			IsFinalBuffer(buffer_id bufID) const;

/* Get attributes of a particular subscriber */
		int32			CountBuffersHeld(subscriber_id subID);

/* Queries for the BBufferStream */
		int32			CountSubscribers() const;
		int32			CountEnteredSubscribers() const;

		subscriber_id	FirstSubscriber() const;
		subscriber_id	LastSubscriber() const;
		subscriber_id	NextSubscriber(subscriber_id subID);
		subscriber_id	PrevSubscriber(subscriber_id subID);

/* debugging aids */
		void			PrintStream();
		void			PrintBuffers();
		void			PrintSubscribers();

/* gaining exclusive access to the BBufferStream */
		bool 			Lock();
		void			Unlock();

/* introduce a new buffer into the "newest" end of the chain */
		status_t		AddBuffer(buffer_id bufID);

/* remove a buffer from the "oldest" end of the chain */
		buffer_id		RemoveBuffer(bool force);

/* allocate a buffer from shared memory and create a bufID for it. */
		buffer_id		CreateBuffer(size_t size, bool isFinal);

/* deallocate a buffer and returns its bufID to the freelist */
		void			DestroyBuffer(buffer_id bufID);

/* remove and destroy any "newest" buffers from the head of the chain
 * that have not yet been claimed by any subscribers.  If there are
 * no subscribers, this clears the entire chain.
 */
		void			RescindBuffers();

/* ================
   Private member functions that assume locking already has been done.
   ================ */

private:

virtual	void		_ReservedBufferStream1();
virtual	void		_ReservedBufferStream2();
virtual	void		_ReservedBufferStream3();
virtual	void		_ReservedBufferStream4();

/* initialize the free list of subscribers */
		void			InitSubscribers();

/* return TRUE if subID appears valid */
		bool			IsSubscribedSafe(subscriber_id subID) const;

/* return TRUE if subID is entered into the stream */
		bool			IsEnteredSafe(subscriber_id subID) const;

/* initialize the free list of buffer IDs */
		void			InitBuffers();

/* Wake a blocked subscriber */
		status_t		WakeSubscriber(subscriber_id subID);

/* Give subID all the buffers it can get */
		void			InheritBuffers(subscriber_id subID);

/* Relinquish any buffers held by subID */
		void			BequeathBuffers(subscriber_id subID);

/* Fast version of ReleaseBuffer() */
		status_t		ReleaseBufferSafe(subscriber_id subID);

/* Release a buffer to a subscriber */
		status_t		ReleaseBufferTo(buffer_id bufID, subscriber_id subID);

/* deallocate all buffers */
		void			FreeAllBuffers();

/* deallocate all subscribers */
		void			FreeAllSubscribers();

/* ================
   Private data members
   ================ */

		BLocker				fLock;
		area_id				fAreaID;		/* area id for this BBufferStream */
		BBufferStreamManager	*fStreamManager;
		BSubscriber			*fHeadFeeder;
		BSubscriber			*fTailFeeder;
		size_t				fHeaderSize;

		/* ================
		   subscribers
		   ================ */

		_sub_info			*fFreeSubs;		/* free list of subscribers */
		_sub_info			*fFirstSub;		/* first entered in itinierary */
		_sub_info			*fLastSub;		/* last entered in itinerary */

		sem_id				fFirstSem;		/* semaphore used by fFirstSub */
		int32				fSubCount;
		int32				fEnteredSubCount;

		_sub_info			fSubscribers[B_MAX_SUBSCRIBER_COUNT];

		/* ================
		   buffers
		   ================ */

		_sbuf_info			*fFreeBuffers;
		_sbuf_info			*fOldestBuffer;	/* first in line */
		_sbuf_info			*fNewestBuffer;	/* fNewest->fNext = NULL */
		int32				fCountBuffers;

		_sbuf_info			fBuffers[B_MAX_BUFFER_COUNT];

		uint32				_reserved[4];
};

#endif 	// #ifdef _BUFFER_STREAM_H