Protect correctly packet reception

kernel/include/conc/deferred_unique_semaphore.hpp

@@ -0,0 +1,100 @@
+//=======================================================================
+// Copyright Baptiste Wicht 2013-2016.
+// Distributed under the terms of the MIT License.
+// (See accompanying file LICENSE or copy at
+//  http://www.opensource.org/licenses/MIT)
+//=======================================================================
+
+#ifndef DEFERRED_UNIQUE_SEMAPHORE_H
+#define DEFERRED_UNIQUE_SEMAPHORE_H
+
+#include "conc/int_lock.hpp"
+
+#include "scheduler.hpp"
+
+/*!
+ * \brief A deferred unique semaphore lock implementation.
+ *
+ * This must be used when the notifier is an IRQ handler.
+ *
+ * Once the lock is acquired, the critical section is only accessible by the
+ * thread who acquired the mutex.
+ */
+struct deferred_unique_semaphore {
+    /*!
+     * \brief Claim the mutex
+     */
+    void claim() {
+        this->pid = scheduler::get_pid();
+    }
+
+    /*!
+     * \brief Wait for the lock
+     */
+    void wait() {
+        int_lock lock;
+
+        lock.lock();
+
+        if (value > 0) {
+            --value;
+
+            lock.unlock();
+        } else {
+            waiting = true;
+
+            scheduler::block_process_light(pid);
+            lock.unlock();
+            scheduler::reschedule();
+
+            waiting = false;
+        }
+    }
+
+    /*!
+     * \brief Release the lock, from an IRQ handler.
+     */
+    void notify() {
+        if (!waiting) {
+            ++value;
+        } else {
+            scheduler::unblock_process_hint(pid);
+        }
+    }
+
+    /*!
+     * \brief Release the lock, from a preeemptible process
+     */
+    void pt_notify() {
+        direct_int_lock lock;
+
+        if (!waiting) {
+            ++value;
+        } else {
+            scheduler::unblock_process_hint(pid);
+        }
+    }
+
+    /*!
+     * \brief Release the lock several times, from an IRQ
+     */
+    void notify(size_t n) {
+        if (!waiting) {
+            value += n;
+        } else {
+            scheduler::unblock_process_hint(pid);
+            --n;
+
+            if (n) {
+                value += n;
+            }
+        }
+    }
+
+private:
+    size_t pid            = 0;     ///< The claimed pid
+    volatile size_t value = 0;     ///< The value of the mutex
+    volatile bool waiting = false; ///< Indicates if the process is waiting
+};
+
+#endif

kernel/include/net/network.hpp

@@ -8,9 +8,6 @@
 #ifndef NETWORK_H
 #define NETWORK_H
 
-#include <conc/mutex.hpp>
-#include <conc/semaphore.hpp>
-
 #include <types.hpp>
 #include <string.hpp>
 #include <circular_buffer.hpp>
@@ -18,6 +15,10 @@
 #include <tuple.hpp>
 #include <expected.hpp>
 
+#include <conc/mutex.hpp>
+#include <conc/semaphore.hpp>
+#include <conc/deferred_unique_semaphore.hpp>
+
 #include "tlib/net_constants.hpp"
 
 #include "net/ethernet_packet.hpp"
@@ -40,9 +41,9 @@ struct interface_descriptor {
     network::ip::address ip_address; ///< The interface IP address
     network::ip::address gateway;    ///< The interface IP gateway
 
-    mutable mutex tx_lock; ///< Mutex protecting the queues
-    mutable semaphore tx_sem; ///< Semaphore for transmission
-    mutable semaphore rx_sem; ///< Semaphore for reception
+    mutable mutex tx_lock;                    ///< Mutex protecting the queues
+    mutable semaphore tx_sem;                 ///< Semaphore for transmission
+    mutable deferred_unique_semaphore rx_sem; ///< Semaphore for reception
 
     size_t rx_thread_pid; ///< The pid of the rx thread
     size_t tx_thread_pid; ///< The pid of the tx thread

kernel/src/drivers/loopback.cpp

@@ -30,7 +30,7 @@ void send_packet(network::interface_descriptor& interface, network::ethernet::pa
     std::copy_n(packet.payload, packet.payload_size, packet_buffer);
 
     interface.rx_queue.emplace_push(packet_buffer, packet.payload_size);
-    interface.rx_sem.unlock();
+    interface.rx_sem.pt_notify();
 
     logging::logf(logging::log_level::TRACE, "loopback: Packet transmitted correctly\n");
 }

kernel/src/drivers/rtl8139.cpp

@@ -130,7 +130,7 @@ void packet_handler(interrupt::syscall_regs*, void* data){
                 network::ethernet::packet packet(packet_buffer, packet_only_length);
 
                 interface.rx_queue.push(packet);
-                interface.rx_sem.irq_unlock();
+                interface.rx_sem.notify();
             }
 
             cur_rx = (cur_rx + packet_length + 4 + 3) & ~3; //align on 4 bytes
@@ -196,7 +196,7 @@ void send_packet(network::interface_descriptor& interface, network::ethernet::pa
         direct_int_lock lock;
 
         interface.rx_queue.emplace_push(packet_buffer, packet.payload_size);
-        interface.rx_sem.unlock();
+        interface.rx_sem.pt_notify();
 
         logging::logf(logging::log_level::TRACE, "rtl8139: Packet to self transmitted correctly\n");
 

kernel/src/net/network.cpp

@@ -53,10 +53,12 @@ void rx_thread(void* data){
 
     auto pid = scheduler::get_pid();
 
+    interface.rx_sem.claim();
+
     logging::logf(logging::log_level::TRACE, "network: RX Thread for interface %u started (pid:%u)\n", interface.id, pid);
 
     while(true){
-        interface.rx_sem.lock();
+        interface.rx_sem.wait();
 
         auto packet = interface.rx_queue.pop();
         network::ethernet::decode(interface, packet);
@@ -158,7 +160,6 @@ void network::init(){
 
                 interface.tx_lock.init(1);
                 interface.tx_sem.init(0);
-                interface.rx_sem.init(0);
             }
 
             sysfs_publish(interface);
@@ -181,7 +182,6 @@ void network::init(){
 
     interface.tx_lock.init(1);
     interface.tx_sem.init(0);
-    interface.rx_sem.init(0);
 
     loopback::init_driver(interface);