diff --git a/headers/private/net/net_buffer.h b/headers/private/net/net_buffer.h index 406121a54b..f269f37f28 100644 --- a/headers/private/net/net_buffer.h +++ b/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; diff --git a/src/add-ons/kernel/network/protocols/udp/udp.cpp b/src/add-ons/kernel/network/protocols/udp/udp.cpp index 1513c277ce..2c727faef1 100644 --- a/src/add-ons/kernel/network/protocols/udp/udp.cpp +++ b/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; } diff --git a/src/add-ons/kernel/network/stack/datalink.cpp b/src/add-ons/kernel/network/stack/datalink.cpp index 4f3d27e5ce..1963c855fa 100644 --- a/src/add-ons/kernel/network/stack/datalink.cpp +++ b/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 #include @@ -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) { diff --git a/src/add-ons/kernel/network/stack/interfaces.cpp b/src/add-ons/kernel/network/stack/interfaces.cpp index e26ee903c1..6f6e2687db 100644 --- a/src/add-ons/kernel/network/stack/interfaces.cpp +++ b/src/add-ons/kernel/network/stack/interfaces.cpp @@ -21,6 +21,7 @@ #include #include +#include #include #include @@ -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); diff --git a/src/add-ons/kernel/network/stack/interfaces.h b/src/add-ons/kernel/network/stack/interfaces.h index bca96f5794..9bcf8fae27 100644 --- a/src/add-ons/kernel/network/stack/interfaces.h +++ b/src/add-ons/kernel/network/stack/interfaces.h @@ -27,7 +27,6 @@ typedef DoublyLinkedList > DeviceMonitorList; struct net_device_interface : DoublyLinkedListLinkImpl { - //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 { DeviceHandlerList receive_funcs; recursive_lock rx_lock; + + thread_id consumer_thread; + net_fifo receive_queue; }; typedef DoublyLinkedList DeviceInterfaceList; diff --git a/src/add-ons/kernel/network/stack/net_buffer.cpp b/src/add-ons/kernel/network/stack/net_buffer.cpp index 407b16c3aa..c317355145 100644 --- a/src/add-ons/kernel/network/stack/net_buffer.cpp +++ b/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; }