net_interface_private: added receive queue and splited device reading from packet processing. Delivering to self no longer is executed in the sender's context, which had some problems with TCP's locking.

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

headers/private/net/net_buffer.h

@@ -23,6 +23,7 @@ typedef struct net_buffer {
 	struct sockaddr *source;
 	struct sockaddr *destination;
 	struct net_interface *interface;
+	int32 type;
 	union {
 		struct {
 			uint16	start;

src/add-ons/kernel/network/protocols/udp/udp.cpp

@@ -654,7 +654,8 @@ UdpEndpointManager::ReceiveData(net_buffer *buffer)
 		return B_ERROR;
 	}
 
-	return B_ERROR;
+	gBufferModule->free(buffer);
+	return B_OK;
 }
 
 

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

@@ -13,6 +13,7 @@
 #include "interfaces.h"
 #include "routes.h"
 #include "stack_private.h"
+#include "utility.h"
 
 #include <net_device.h>
 #include <KernelExport.h>
@@ -54,41 +55,22 @@ device_reader_thread(void *_interface)
 		status = device->module->receive_data(device, &buffer);
 		rx_lock.Lock();
 		if (status == B_OK) {
-			//dprintf("received buffer of %ld bytes length\n", buffer->size);
-
 			// feed device monitors
-			DeviceMonitorList::Iterator iterator
-				= interface->monitor_funcs.GetIterator();
+			DeviceMonitorList::Iterator iterator =
+				interface->monitor_funcs.GetIterator();
 			while (iterator.HasNext()) {
 				net_device_monitor *monitor = iterator.Next();
 				monitor->receive(monitor, buffer);
 			}
 
-			int32 type = interface->deframe_func(device, buffer);
-			if (type >= 0) {
-				// find handler for this packet
-				DeviceHandlerList::Iterator iterator
-					= interface->receive_funcs.GetIterator();
-				status = B_ERROR;
-
-				while (iterator.HasNext()) {
-					net_device_handler *handler = iterator.Next();
-
-					if (handler->type == type) {
-						status = handler->func(handler->cookie, device, buffer);
-						if (status == B_OK)
-							break;
-					}
-				}
-			} else
-				status = type;
-
-			if (status == B_OK) {
-				// the buffer no longer belongs to us
+			buffer->interface = NULL;
+			buffer->type = interface->deframe_func(interface->device, buffer);
+			if (buffer->type < 0) {
+				gNetBufferModule.free(buffer);
 				continue;
 			}
 
-			gNetBufferModule.free(buffer);
+			fifo_enqueue_buffer(&interface->receive_queue, buffer);
 		} else {
 			// In case of error, give the other threads some
 			// time to run since this is a near real time thread.
@@ -336,8 +318,8 @@ datalink_control(net_domain *_domain, int32 option, void *value,
 status_t
 datalink_send_data(struct net_route *route, net_buffer *buffer)
 {
-	net_interface *interface = route->interface;
-	net_domain *domain = interface->domain;
+	net_interface_private *interface =
+		(net_interface_private *)route->interface;
 
 	//dprintf("send buffer (%ld bytes) to interface %s (route flags %lx)\n",
 	//	buffer->size, interface->name, route->flags);
@@ -346,8 +328,12 @@ datalink_send_data(struct net_route *route, net_buffer *buffer)
 		return ENETUNREACH;
 
 	if (route->flags & RTF_LOCAL) {
+		// we set the interface here so the buffer is delivered directly
+		// to the domain in interfaces.cpp:device_consumer_thread()
+		buffer->interface = interface;
 		// this one goes back to the domain directly
-		return domain->module->receive_data(buffer);
+		return fifo_enqueue_buffer(
+			&interface->device_interface->receive_queue, buffer);
 	}
 
 	if (route->flags & RTF_GATEWAY) {

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

@@ -21,6 +21,7 @@
 
 #include <net/if_dl.h>
 #include <new>
+#include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 
@@ -39,6 +40,51 @@ static uint32 sInterfaceIndex;
 static uint32 sDeviceIndex;
 
 
+static status_t
+device_consumer_thread(void *_interface)
+{
+	net_device_interface *interface = (net_device_interface *)_interface;
+	net_device *device = interface->device;
+	net_buffer *buffer;
+
+	while (true) {
+		ssize_t status = fifo_dequeue_buffer(&interface->receive_queue, 0,
+			B_INFINITE_TIMEOUT, &buffer);
+		if (status == B_INTERRUPTED)
+			continue;
+		else if (status < B_OK)
+			break;
+
+		if (buffer->interface != NULL) {
+			// if the interface is already specified this buffer was
+			// delivered locally.
+
+			net_domain *domain = buffer->interface->domain;
+
+			if (domain->module->receive_data(buffer) == B_OK)
+				buffer = NULL;
+		} else {
+			// find handler for this packet
+			DeviceHandlerList::Iterator it2 =
+				interface->receive_funcs.GetIterator();
+			while (buffer && it2.HasNext()) {
+				net_device_handler *handler = it2.Next();
+
+				// if the handler returns B_OK, it consumed the buffer
+				if (handler->type == buffer->type
+					&& handler->func(handler->cookie, device, buffer) == B_OK)
+					buffer = NULL;
+			}
+		}
+
+		if (buffer)
+			gNetBufferModule.free(buffer);
+	}
+
+	return B_OK;
+}
+
+
 static net_device_interface *
 find_device_interface(const char *name)
 {
@@ -65,6 +111,60 @@ domain_receive_adapter(void *cookie, net_device *device, net_buffer *buffer)
 }
 
 
+static net_device_interface *
+allocate_device_interface(net_device *device, net_device_module_info *module)
+{
+	net_device_interface *interface = new (std::nothrow) net_device_interface;
+	if (interface == NULL)
+		goto error_0;
+
+	if (recursive_lock_init(&interface->rx_lock, "rx lock") < B_OK)
+		goto error_1;
+
+	char name[128];
+	snprintf(name, sizeof(name), "%s receive queue", device->name);
+
+	if (init_fifo(&interface->receive_queue, name, 16 * 1024 * 1024) < B_OK)
+		goto error_2;
+
+	interface->name = device->name;
+	interface->module = module;
+	interface->device = device;
+	interface->up_count = 0;
+	interface->ref_count = 1;
+	interface->deframe_func = NULL;
+	interface->deframe_ref_count = 0;
+
+	snprintf(name, sizeof(name), "%s consumer", device->name);
+
+	interface->reader_thread   = -1;
+	interface->consumer_thread = spawn_kernel_thread(device_consumer_thread,
+		name, B_REAL_TIME_DISPLAY_PRIORITY - 10, interface);
+	if (interface->consumer_thread < B_OK)
+		goto error_3;
+	resume_thread(interface->consumer_thread);
+
+	// TODO: proper interface index allocation
+	device->index = ++sDeviceIndex;
+	device->module = module;
+
+	sInterfaces.Add(interface);
+	return interface;
+
+error_3:
+	uninit_fifo(&interface->receive_queue);
+
+error_2:
+	recursive_lock_destroy(&interface->rx_lock);
+
+error_1:
+	delete interface;
+
+error_0:
+	return NULL;
+}
+
+
 net_device_interface *
 grab_device_interface(net_device_interface *interface)
 {
@@ -337,13 +437,15 @@ put_device_interface(struct net_device_interface *interface)
 	if (atomic_add(&interface->ref_count, -1) != 1)
 		return;
 
-	// we need to remove this interface!
-
 	{
 		BenaphoreLocker locker(sInterfaceLock);
 		sInterfaces.Remove(interface);
 	}
 
+	uninit_fifo(&interface->receive_queue);
+	status_t status;
+	wait_for_thread(interface->consumer_thread, &status);
+
 	interface->module->uninit_device(interface->device);
 	put_module(interface->module->info.name);
 
@@ -411,27 +513,9 @@ get_device_interface(const char *name, bool create)
 			net_device *device;
 			status_t status = module->init_device(name, &device);
 			if (status == B_OK) {
-				// create new module interface for this
-				interface = new (std::nothrow) net_device_interface;
-				if (interface != NULL) {
-					if (recursive_lock_init(&interface->rx_lock,
-							"rx lock") >= B_OK) {
-						interface->name = device->name;
-						interface->module = module;
-						interface->device = device;
-						interface->up_count = 0;
-						interface->ref_count = 1;
-						interface->deframe_func = NULL;
-						interface->deframe_ref_count = 0;
-
-						device->index = ++sDeviceIndex;
-						device->module = module;
-
-						sInterfaces.Add(interface);
-						return interface;
-					}
-					delete interface;
-				}
+				interface = allocate_device_interface(device, module);
+				if (interface)
+					return interface;
 				module->uninit_device(device);
 			}
 			put_module(moduleName);
@@ -469,6 +553,8 @@ down_device_interface(net_device_interface *interface)
 
 	thread_id reader_thread = interface->reader_thread;
 
+	// TODO when setting the interface down, should we clear the receive queue?
+
 	// one of the callers must hold a reference to the net_device_interface
 	// usually it is one of the net_interfaces.
 	recursive_lock_unlock(&interface->rx_lock);

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

@@ -27,7 +27,6 @@ typedef DoublyLinkedList<net_device_monitor,
 	DoublyLinkedListCLink<net_device_monitor> > DeviceMonitorList;
 
 struct net_device_interface : DoublyLinkedListLinkImpl<net_device_interface> {
-	//struct list_link	link;
 	const char			*name;
 	struct net_device_module_info *module;
 	struct net_device	*device;
@@ -43,6 +42,9 @@ struct net_device_interface : DoublyLinkedListLinkImpl<net_device_interface> {
 	DeviceHandlerList	receive_funcs;
 
 	recursive_lock		rx_lock;
+
+	thread_id			consumer_thread;
+	net_fifo			receive_queue;
 };
 
 typedef DoublyLinkedList<net_device_interface> DeviceInterfaceList;

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

@@ -345,6 +345,8 @@ create_buffer(size_t headerSpace)
 	buffer->flags = 0;
 	buffer->size = 0;
 
+	buffer->type = -1;
+
 	return buffer;
 }
 
@@ -378,6 +380,7 @@ copy_metadata(net_buffer *destination, const net_buffer *source)
 	destination->offset = source->offset;
 	destination->size = source->size;
 	destination->protocol = source->protocol;
+	destination->type = source->type;
 }