Prepared net_device_monitor to accept device removal events.

- Introduced public net_device_monitor.
 - Changed the link protocol to maintain a lock per instance instead of inside the FIFO. Now all of the link instance data is protected.
 - Adapted the link protocol to use net_device_monitor.
 - Introduced a private Fifo class which doesn't maintain it's own lock.
  - Maybe we should add something like a public net_protocol_implementation which maintains a fifo and a benaphore? With the fifo using the structure's lock instead of maintaining it's own.


git-svn-id: file:///srv/svn/repos/haiku/haiku/trunk@20614 a95241bf-73f2-0310-859d-f6bbb57e9c96

headers/private/net/net_stack.h

@@ -37,6 +37,21 @@ struct net_timer {
 typedef int32 (*net_deframe_func)(struct net_device *device, struct net_buffer *buffer);
 typedef status_t (*net_receive_func)(void *cookie, struct net_buffer *buffer);
 
+enum {
+	B_DEVICE_GOING_UP = 1,
+	B_DEVICE_GOING_DOWN,
+	B_DEVICE_BEING_REMOVED,
+};
+
+struct net_device_monitor {
+	struct list_link link;
+	void *cookie;
+
+	status_t (*receive)(struct net_device_monitor *monitor,
+		struct net_buffer *buffer);
+	void (*event)(struct net_device_monitor *monitor, int32 event);
+};
+
 struct net_stack_module_info {
 	module_info info;
 
@@ -68,9 +83,9 @@ struct net_stack_module_info {
 	status_t (*unregister_device_handler)(struct net_device *device, int32 type);
 
 	status_t (*register_device_monitor)(struct net_device *device,
-					net_receive_func receiveFunc, void *cookie);
+		struct net_device_monitor *monitor);
 	status_t (*unregister_device_monitor)(struct net_device *device,
-					net_receive_func receiveFunc, void *cookie);
+		struct net_device_monitor *monitor);
 
 	status_t (*device_link_changed)(struct net_device *device);
 	status_t (*device_removed)(struct net_device *device);

src/add-ons/kernel/network/stack/datalink.cpp

@@ -60,7 +60,7 @@ device_reader_thread(void *_interface)
 				= interface->monitor_funcs.GetIterator();
 			while (iterator.HasNext()) {
 				net_device_monitor *monitor = iterator.Next();
-				monitor->func(monitor->cookie, buffer);
+				monitor->receive(monitor, buffer);
 			}
 
 			int32 type = interface->deframe_func(device, buffer);
@@ -478,7 +478,7 @@ interface_protocol_send_data(net_datalink_protocol *_protocol,
 		interface->device_interface->monitor_funcs.GetIterator();
 	while (iterator.HasNext()) {
 		net_device_monitor *monitor = iterator.Next();
-		monitor->func(monitor->cookie, buffer);
+		monitor->receive(monitor, buffer);
 	}
 
 	return protocol->device_module->send_data(protocol->device, buffer);

src/add-ons/kernel/network/stack/interfaces.cpp

@@ -593,9 +593,11 @@ unregister_device_handler(struct net_device *device, int32 type)
 
 
 status_t
-register_device_monitor(struct net_device *device,
-	net_receive_func receiveFunc, void *cookie)
+register_device_monitor(net_device *device, net_device_monitor *monitor)
 {
+	if (monitor->receive == NULL || monitor->event == NULL)
+		return B_BAD_VALUE;
+
 	BenaphoreLocker locker(sInterfaceLock);
 
 	// find device interface for this device
@@ -604,23 +606,13 @@ register_device_monitor(struct net_device *device,
 		return ENODEV;
 
 	BenaphoreLocker _(interface->rx_lock);
-
-	// Add new monitor
-
-	net_device_monitor *monitor = new (std::nothrow) net_device_monitor;
-	if (monitor == NULL)
-		return B_NO_MEMORY;
-
-	monitor->func = receiveFunc;
-	monitor->cookie = cookie;
 	interface->monitor_funcs.Add(monitor);
 	return B_OK;
 }
 
 
 status_t
-unregister_device_monitor(struct net_device *device,
-	net_receive_func receiveFunc, void *cookie)
+unregister_device_monitor(net_device *device, net_device_monitor *monitor)
 {
 	BenaphoreLocker locker(sInterfaceLock);
 
@@ -635,12 +627,8 @@ unregister_device_monitor(struct net_device *device,
 
 	DeviceMonitorList::Iterator iterator = interface->monitor_funcs.GetIterator();
 	while (iterator.HasNext()) {
-		net_device_monitor *monitor = iterator.Next();
-
-		if (monitor->cookie == cookie && monitor->func == receiveFunc) {
-			// found it
+		if (iterator.Next() == monitor) {
 			iterator.Remove();
-			delete monitor;
 			return B_OK;
 		}
 	}

src/add-ons/kernel/network/stack/interfaces.h

@@ -21,13 +21,10 @@ struct net_device_handler : public DoublyLinkedListLinkImpl<net_device_handler>
 	void				*cookie;
 };
 
-struct net_device_monitor : public DoublyLinkedListLinkImpl<net_device_monitor> {
-	net_receive_func	func;
-	void				*cookie;
-};
-
 typedef DoublyLinkedList<net_device_handler> DeviceHandlerList;
-typedef DoublyLinkedList<net_device_monitor> DeviceMonitorList;
+
+typedef DoublyLinkedList<net_device_monitor,
+	DoublyLinkedListCLink<net_device_monitor> > DeviceMonitorList;
 
 struct net_device_interface : DoublyLinkedListLinkImpl<net_device_interface> {
 	//struct list_link	link;
@@ -93,9 +90,9 @@ status_t register_device_handler(struct net_device *device, int32 type,
 	net_receive_func receiveFunc, void *cookie);
 status_t unregister_device_handler(struct net_device *device, int32 type);
 status_t register_device_monitor(struct net_device *device,
-	net_receive_func receiveFunc, void *cookie);
+	struct net_device_monitor *monitor);
 status_t unregister_device_monitor(struct net_device *device,
-	net_receive_func receiveFunc, void *cookie);
+	struct net_device_monitor *monitor);
 status_t device_link_changed(net_device *device);
 status_t device_removed(net_device *device);
 

src/add-ons/kernel/network/stack/link.cpp

@@ -16,6 +16,9 @@
 
 #include <net_device.h>
 
+#include <lock.h>
+#include <util/AutoLock.h>
+
 #include <KernelExport.h>
 
 #include <net/if_types.h>
@@ -25,23 +28,173 @@
 #include <sys/sockio.h>
 
 
-struct link_protocol : net_protocol {
-	net_fifo	fifo;
-	char		registered_interface[IF_NAMESIZE];
-	bool		registered_monitor;
+class LinkProtocol : public net_protocol {
+public:
+	LinkProtocol();
+	~LinkProtocol();
+
+	status_t InitCheck() const;
+
+	status_t StartMonitoring(const char *);
+	status_t StopMonitoring();
+
+	ssize_t ReadData(size_t numBytes, uint32 flags, net_buffer **_buffer);
+	ssize_t ReadAvail() const;
+
+private:
+	status_t _Enqueue(net_buffer *buffer);
+	status_t _Unregister();
+
+	mutable benaphore	fLock;
+	Fifo				fFifo;
+
+	net_device_monitor	fMonitor;
+	net_device_interface *fMonitoredDevice;
+
+	static status_t _MonitorData(net_device_monitor *monitor, net_buffer *buffer);
+	static void _MonitorEvent(net_device_monitor *monitor, int32 event);
 };
 
 
 struct net_domain *sDomain;
 
 
-static status_t
-link_monitor_data(void *cookie, net_buffer *packet)
+LinkProtocol::LinkProtocol()
+	: fFifo("packet monitor fifo", 65536)
 {
-	link_protocol *protocol = (link_protocol *)cookie;
+	benaphore_init(&fLock, "packet monitor lock");
 
-	return fifo_socket_enqueue_buffer(&protocol->fifo, protocol->socket,
-						B_SELECT_READ, packet);
+	fMonitor.cookie = this;
+	fMonitor.receive = _MonitorData;
+	fMonitor.event = _MonitorEvent;
+	fMonitoredDevice = NULL;
+}
+
+
+LinkProtocol::~LinkProtocol()
+{
+	if (fMonitoredDevice) {
+		unregister_device_monitor(fMonitoredDevice->device, &fMonitor);
+		put_device_interface(fMonitoredDevice);
+	}
+
+	benaphore_destroy(&fLock);
+}
+
+
+status_t
+LinkProtocol::InitCheck() const
+{
+	return fLock.sem >= 0 && fFifo.InitCheck();
+}
+
+
+status_t
+LinkProtocol::StartMonitoring(const char *deviceName)
+{
+	BenaphoreLocker _(fLock);
+
+	if (fMonitoredDevice)
+		return B_BUSY;
+
+	net_device_interface *interface = get_device_interface(deviceName);
+	if (interface == NULL)
+		return ENODEV;
+
+	status_t status = register_device_monitor(interface->device, &fMonitor);
+	if (status < B_OK) {
+		put_device_interface(interface);
+		return status;
+	}
+
+	fMonitoredDevice = interface;
+	return B_OK;
+}
+
+
+status_t
+LinkProtocol::StopMonitoring()
+{
+	BenaphoreLocker _(fLock);
+
+	// TODO compare our device with the supplied device name?
+	return _Unregister();
+}
+
+
+ssize_t
+LinkProtocol::ReadData(size_t numBytes, uint32 flags, net_buffer **_buffer)
+{
+	BenaphoreLocker _(fLock);
+
+	if (fMonitoredDevice == NULL) {
+		if (fFifo.current_bytes == 0)
+			return ENODEV;
+	}
+
+	net_buffer *buffer;
+	status_t status = fFifo.Dequeue(&fLock, flags, socket->receive.timeout,
+		&buffer);
+	if (status < B_OK)
+		return status;
+
+	*_buffer = buffer;
+	return B_OK;
+}
+
+
+ssize_t
+LinkProtocol::ReadAvail() const
+{
+	BenaphoreLocker _(fLock);
+	if (fMonitoredDevice == NULL)
+		return ECONNRESET;
+	return fFifo.current_bytes;
+}
+
+
+status_t
+LinkProtocol::_Unregister()
+{
+	if (fMonitoredDevice == NULL)
+		return B_BAD_VALUE;
+
+	status_t status = unregister_device_monitor(fMonitoredDevice->device,
+		&fMonitor);
+	put_device_interface(fMonitoredDevice);
+	fMonitoredDevice = NULL;
+
+	return status;
+}
+
+
+status_t
+LinkProtocol::_Enqueue(net_buffer *buffer)
+{
+	BenaphoreLocker _(fLock);
+	return fFifo.EnqueueAndNotify(buffer, socket, B_SELECT_READ);
+}
+
+
+status_t
+LinkProtocol::_MonitorData(net_device_monitor *monitor, net_buffer *packet)
+{
+	return ((LinkProtocol *)monitor->cookie)->_Enqueue(packet);
+}
+
+
+void
+LinkProtocol::_MonitorEvent(net_device_monitor *monitor, int32 event)
+{
+	LinkProtocol *protocol = (LinkProtocol *)monitor->cookie;
+
+	// We currently maintain the monitor while the device is down
+	if (event == B_DEVICE_BEING_REMOVED) {
+		BenaphoreLocker _(protocol->fLock);
+
+		protocol->_Unregister();
+		notify_socket(protocol->socket, B_SELECT_READ, ECONNRESET);
+	}
 }
 
 
@@ -51,35 +204,20 @@ link_monitor_data(void *cookie, net_buffer *packet)
 net_protocol *
 link_init_protocol(net_socket *socket)
 {
-	link_protocol *protocol = new (std::nothrow) link_protocol;
-	if (protocol == NULL)
-		return NULL;
-
-	if (init_fifo(&protocol->fifo, "packet monitor socket", 65536) < B_OK) {
+	LinkProtocol *protocol = new (std::nothrow) LinkProtocol();
+	if (protocol && protocol->InitCheck() < B_OK) {
 		delete protocol;
 		return NULL;
 	}
 
-	protocol->registered_monitor = false;
 	return protocol;
 }
 
 
 status_t
-link_uninit_protocol(net_protocol *_protocol)
+link_uninit_protocol(net_protocol *protocol)
 {
-	link_protocol *protocol = (link_protocol *)_protocol;
-
-	if (protocol->registered_monitor) {
-		net_device_interface *interface = get_device_interface(protocol->registered_interface);
-		if (interface != NULL) {
-			unregister_device_monitor(interface->device, link_monitor_data, protocol);
-			put_device_interface(interface);
-		}
-	}
-
-	uninit_fifo(&protocol->fifo);
-	delete protocol;
+	delete (LinkProtocol *)protocol;
 	return B_OK;
 }
 
@@ -123,7 +261,9 @@ status_t
 link_control(net_protocol *_protocol, int level, int option, void *value,
 	size_t *_length)
 {
-	link_protocol *protocol = (link_protocol *)_protocol;
+	LinkProtocol *protocol = (LinkProtocol *)_protocol;
+
+	// TODO All of this common functionality should be elsewhere
 
 	switch (option) {
 		case SIOCGIFINDEX:
@@ -203,58 +343,15 @@ link_control(net_protocol *_protocol, int level, int option, void *value,
 
 		case SIOCSPACKETCAP:
 		{
-			// start packet monitoring
-
-			if (protocol->registered_monitor)
-				return B_BUSY;
-
 			struct ifreq request;
 			if (user_memcpy(&request, value, IF_NAMESIZE) < B_OK)
 				return B_BAD_ADDRESS;
 
-			net_device_interface *interface = get_device_interface(request.ifr_name);
-			status_t status;
-			if (interface != NULL) {
-				status = register_device_monitor(interface->device,
-					link_monitor_data, protocol);
-				if (status == B_OK) {
-					// we're now registered
-					strlcpy(protocol->registered_interface, request.ifr_name, IF_NAMESIZE);
-					protocol->registered_monitor = true;
-				}
-				put_device_interface(interface);
-			} else
-				status = ENODEV;
-
-			return status;
+			return protocol->StartMonitoring(request.ifr_name);
 		}
 
 		case SIOCCPACKETCAP:
-		{
-			// stop packet monitoring
-
-			if (!protocol->registered_monitor)
-				return B_BAD_VALUE;
-
-			struct ifreq request;
-			if (user_memcpy(&request, value, IF_NAMESIZE) < B_OK)
-				return B_BAD_ADDRESS;
-
-			net_device_interface *interface = get_device_interface(request.ifr_name);
-			status_t status;
-			if (interface != NULL) {
-				status = unregister_device_monitor(interface->device,
-					link_monitor_data, protocol);
-				if (status == B_OK) {
-					// we're now no longer registered
-					protocol->registered_monitor = false;
-				}
-				put_device_interface(interface);
-			} else
-				status = ENODEV;
-
-			return status;
-		}
+			return protocol->StopMonitoring();
 	}
 
 	return datalink_control(sDomain, option, value, _length);
@@ -313,29 +410,17 @@ link_send_avail(net_protocol *protocol)
 
 
 status_t
-link_read_data(net_protocol *_protocol, size_t numBytes, uint32 flags,
+link_read_data(net_protocol *protocol, size_t numBytes, uint32 flags,
 	net_buffer **_buffer)
 {
-	link_protocol *protocol = (link_protocol *)_protocol;
-
-	dprintf("link_read is waiting for data...\n");
-
-	net_buffer *buffer;
-	status_t status = fifo_dequeue_buffer(&protocol->fifo,
-		flags, protocol->socket->receive.timeout, &buffer);
-	if (status < B_OK)
-		return status;
-
-	*_buffer = buffer;
-	return B_OK;
+	return ((LinkProtocol *)protocol)->ReadData(numBytes, flags, _buffer);
 }
 
 
 ssize_t
-link_read_avail(net_protocol *_protocol)
+link_read_avail(net_protocol *protocol)
 {
-	link_protocol *protocol = (link_protocol *)_protocol;
-	return protocol->fifo.current_bytes;
+	return ((LinkProtocol *)protocol)->ReadAvail();
 }
 
 

src/add-ons/kernel/network/stack/utility.cpp

@@ -24,6 +24,126 @@ static thread_id sTimerThread;
 static bigtime_t sTimerTimeout;
 
 
+template<typename FifoType> static inline status_t
+base_fifo_init(FifoType *fifo, const char *name, size_t maxBytes)
+{
+	fifo->notify = create_sem(1, name);
+	fifo->max_bytes = maxBytes;
+	fifo->current_bytes = 0;
+	fifo->waiting = 0;
+	list_init(&fifo->buffers);
+
+	return fifo->notify;
+}
+
+
+template<typename FifoType> static inline status_t
+base_fifo_enqueue_buffer(FifoType *fifo, net_buffer *buffer)
+{
+	if (fifo->max_bytes > 0 && fifo->current_bytes + buffer->size > fifo->max_bytes)
+		return ENOBUFS;
+
+	list_add_item(&fifo->buffers, buffer);
+	fifo->current_bytes += buffer->size;
+
+	if (fifo->waiting > 0) {
+		fifo->waiting--;
+		release_sem_etc(fifo->notify, 1, B_DO_NOT_RESCHEDULE);
+			// we still hold the benaphore lock, so it makes no sense
+			// to reschedule after having released the sync semaphore
+	}
+
+	return B_OK;
+}
+
+
+/*!
+	Gets the first buffer from the FIFO. If there is no buffer, it
+	will wait depending on the \a flags and \a timeout.
+	The following flags are supported (the rest is ignored):
+		MSG_DONTWAIT - ignores the timeout and never wait for a buffer; if your
+			socket is O_NONBLOCK, you should specify this flag. A \a timeout of
+			zero is equivalent to this flag, though.
+		MSG_PEEK - returns a clone of the buffer and keep the original
+			in the FIFO.
+*/
+template<typename FifoType> static inline ssize_t
+base_fifo_dequeue_buffer(FifoType *fifo, benaphore *lock, uint32 flags,
+	bigtime_t timeout, net_buffer **_buffer)
+{
+	// this function is called with `lock' held.
+	bool dontWait = (flags & MSG_DONTWAIT) != 0 || timeout == 0;
+	status_t status;
+
+	while (true) {
+		net_buffer *buffer = (net_buffer *)list_get_first_item(&fifo->buffers);
+		if (buffer != NULL) {
+			if ((flags & MSG_PEEK) != 0) {
+				// we need to clone the buffer for inspection; we can't give a
+				// handle to a buffer that we're still using
+				buffer = gNetBufferModule.clone(buffer, false);
+				if (buffer == NULL) {
+					status = B_NO_MEMORY;
+					break;
+				}
+			} else {
+				list_remove_item(&fifo->buffers, buffer);
+				fifo->current_bytes -= buffer->size;
+			}
+
+			*_buffer = buffer;
+			status = B_OK;
+			break;
+		}
+
+		if (!dontWait)
+			fifo->waiting++;
+
+		// we need to wait until a new buffer becomes available
+		benaphore_unlock(lock);
+
+		if (dontWait)
+			return B_WOULD_BLOCK;
+
+		status = acquire_sem_etc(fifo->notify, 1,
+			B_CAN_INTERRUPT | B_RELATIVE_TIMEOUT, timeout);
+		if (status < B_OK)
+			return status;
+
+		// try again
+		benaphore_lock(lock);
+	}
+
+	if ((flags & MSG_PEEK) != 0 && fifo->waiting > 0) {
+		// another thread is waiting for data, since we didn't eat the
+		// buffer, it gets it
+		fifo->waiting--;
+		release_sem_etc(fifo->notify, 1, B_DO_NOT_RESCHEDULE);
+	}
+
+	return status;
+}
+
+
+template<typename FifoType> static inline status_t
+base_fifo_clear(FifoType *fifo)
+{
+	while (true) {
+		net_buffer *buffer = (net_buffer *)list_remove_head_item(&fifo->buffers);
+		if (buffer == NULL)
+			break;
+
+		gNetBufferModule.free(buffer);
+	}
+
+	fifo->current_bytes = 0;
+	return B_OK;
+}
+
+
+//	#pragma mark -
+
+
 void *
 UserBuffer::Copy(void *source, size_t length)
 {
@@ -101,6 +221,65 @@ notify_socket(net_socket *socket, uint8 event, int32 value)
 //	#pragma mark - FIFOs
 
 
+Fifo::Fifo(const char *name, size_t maxBytes)
+{
+	base_fifo_init(this, name, maxBytes);
+}
+
+
+Fifo::~Fifo()
+{
+	Clear();
+	delete_sem(notify);
+}
+
+
+status_t
+Fifo::InitCheck() const
+{
+	return !(notify < B_OK);
+}
+
+
+status_t
+Fifo::Enqueue(net_buffer *buffer)
+{
+	return base_fifo_enqueue_buffer(this, buffer);
+}
+
+
+status_t
+Fifo::EnqueueAndNotify(net_buffer *_buffer, net_socket *socket, uint8 event)
+{
+	net_buffer *buffer = gNetBufferModule.clone(_buffer, false);
+	if (buffer == NULL)
+		return B_NO_MEMORY;
+
+	status_t status = Enqueue(buffer);
+	if (status < B_OK)
+		gNetBufferModule.free(buffer);
+	else
+		notify_socket(socket, event, current_bytes);
+
+	return status;
+}
+
+
+ssize_t
+Fifo::Dequeue(benaphore *lock, uint32 flags, bigtime_t timeout,
+	net_buffer **_buffer)
+{
+	return base_fifo_dequeue_buffer(this, lock, flags, timeout, _buffer);
+}
+
+
+ssize_t
+Fifo::Clear()
+{
+	return base_fifo_clear(this);
+}
+
+
 status_t
 init_fifo(net_fifo *fifo, const char *name, size_t maxBytes)
 {
@@ -108,19 +287,11 @@ init_fifo(net_fifo *fifo, const char *name, size_t maxBytes)
 	if (status < B_OK)
 		return status;
 
-	fifo->notify = create_sem(1, name);
-	if (fifo->notify < B_OK) {
+	status = base_fifo_init(fifo, name, maxBytes);
+	if (status < B_OK)
 		benaphore_destroy(&fifo->lock);
-		return fifo->notify;
-	}
 
-	fifo->max_bytes = maxBytes;
-	fifo->current_bytes = 0;
-	fifo->waiting = 0;
-
-	list_init(&fifo->buffers);
-
-	return B_OK;
+	return status;
 }
 
 
@@ -134,99 +305,19 @@ uninit_fifo(net_fifo *fifo)
 }
 
 
-static status_t
-_fifo_enqueue_buffer(net_fifo *fifo, net_buffer *buffer)
-{
-	if (fifo->max_bytes > 0 && fifo->current_bytes + buffer->size > fifo->max_bytes)
-		return ENOBUFS;
-
-	list_add_item(&fifo->buffers, buffer);
-	fifo->current_bytes += buffer->size;
-
-	if (fifo->waiting > 0) {
-		fifo->waiting--;
-		release_sem_etc(fifo->notify, 1, B_DO_NOT_RESCHEDULE);
-			// we still hold the benaphore lock, so it makes no sense
-			// to reschedule after having released the sync semaphore
-	}
-
-	return B_OK;
-}
-
 status_t
 fifo_enqueue_buffer(net_fifo *fifo, net_buffer *buffer)
 {
 	BenaphoreLocker locker(fifo->lock);
-	return _fifo_enqueue_buffer(fifo, buffer);
+	return base_fifo_enqueue_buffer(fifo, buffer);
 }
 
 
-/*!
-	Gets the first buffer from the FIFO. If there is no buffer, it
-	will wait depending on the \a flags and \a timeout.
-	The following flags are supported (the rest is ignored):
-		MSG_DONTWAIT - ignores the timeout and never wait for a buffer; if your
-			socket is O_NONBLOCK, you should specify this flag. A \a timeout of
-			zero is equivalent to this flag, though.
-		MSG_PEEK - returns a clone of the buffer and keep the original
-			in the FIFO.
-*/
-ssize_t
-fifo_dequeue_buffer(net_fifo *fifo, uint32 flags, bigtime_t timeout,
-	net_buffer **_buffer)
+ssize_t fifo_dequeue_buffer(net_fifo *fifo, uint32 flags, bigtime_t timeout,
+	struct net_buffer **_buffer)
 {
-	benaphore_lock(&fifo->lock);
-	bool dontWait = (flags & MSG_DONTWAIT) != 0 || timeout == 0;
-	status_t status;
-
-	while (true) {
-		net_buffer *buffer = (net_buffer *)list_get_first_item(&fifo->buffers);
-		if (buffer != NULL) {
-			if ((flags & MSG_PEEK) != 0) {
-				// we need to clone the buffer for inspection; we can't give a
-				// handle to a buffer that we're still using
-				buffer = gNetBufferModule.clone(buffer, false);
-				if (buffer == NULL) {
-					status = B_NO_MEMORY;
-					break;
-				}
-			} else {
-				list_remove_item(&fifo->buffers, buffer);
-				fifo->current_bytes -= buffer->size;
-			}
-
-			*_buffer = buffer;
-			status = B_OK;
-			break;
-		}
-
-		if (!dontWait)
-			fifo->waiting++;
-
-		// we need to wait until a new buffer becomes available
-		benaphore_unlock(&fifo->lock);
-
-		if (dontWait)
-			return B_WOULD_BLOCK;
-
-		status = acquire_sem_etc(fifo->notify, 1,
-			B_CAN_INTERRUPT | B_RELATIVE_TIMEOUT, timeout);
-		if (status < B_OK)
-			return status;
-
-		// try again
-		benaphore_lock(&fifo->lock);
-	}
-
-	if ((flags & MSG_PEEK) != 0 && fifo->waiting > 0) {
-		// another thread is waiting for data, since we didn't eat the
-		// buffer, it gets it
-		fifo->waiting--;
-		release_sem_etc(fifo->notify, 1, B_DO_NOT_RESCHEDULE);
-	}
-
-	benaphore_unlock(&fifo->lock);
-	return status;
+	BenaphoreLocker locker(fifo->lock);
+	return base_fifo_dequeue_buffer(fifo, &fifo->lock, flags, timeout, _buffer);
 }
 
 
@@ -234,17 +325,7 @@ status_t
 clear_fifo(net_fifo *fifo)
 {
 	BenaphoreLocker locker(fifo->lock);
-
-	while (true) {
-		net_buffer *buffer = (net_buffer *)list_remove_head_item(&fifo->buffers);
-		if (buffer == NULL)
-			break;
-
-		gNetBufferModule.free(buffer);
-	}
-
-	fifo->current_bytes = 0;
-	return B_OK;
+	return base_fifo_clear(fifo);
 }
 
 
@@ -258,7 +339,7 @@ fifo_socket_enqueue_buffer(net_fifo *fifo, net_socket *socket, uint8 event,
 
 	BenaphoreLocker locker(fifo->lock);
 
-	status_t status = _fifo_enqueue_buffer(fifo, buffer);
+	status_t status = base_fifo_enqueue_buffer(fifo, buffer);
 	if (status < B_OK)
 		gNetBufferModule.free(buffer);
 	else

src/add-ons/kernel/network/stack/utility.h

@@ -26,6 +26,33 @@ private:
 };
 
 
+// internal Fifo class which doesn't maintain it's own lock
+class Fifo {
+public:
+	Fifo(const char *name, size_t maxBytes);
+	~Fifo();
+
+	status_t InitCheck() const;
+
+	status_t Enqueue(net_buffer *buffer);
+	status_t EnqueueAndNotify(net_buffer *_buffer, net_socket *socket, uint8 event);
+
+	ssize_t Dequeue(benaphore *lock, uint32 flags, bigtime_t timeout,
+		net_buffer **_buffer);
+	status_t Clear();
+
+	bool IsEmpty() const { return current_bytes == 0; }
+
+//private:
+	// these field names are kept so we can use templatized
+	// functions together with net_fifo
+	sem_id		notify;
+	int32		waiting;
+	size_t		max_bytes;
+	size_t		current_bytes;
+	struct list	buffers;
+};
+
 inline
 UserBuffer::UserBuffer(void *buffer, size_t size)
 	: fBuffer((uint8 *)buffer), fBufferSize(size),