Merge branch 'fastclick/threadvector' into queuedevice, allows to...

Merge branch 'fastclick/threadvector' into queuedevice, allows to automatically detect if threads end up in the output elements

elements/userlevel/queuedevice.cc

@@ -59,7 +59,6 @@ int QueueDevice::configure_rx(int numa_node,unsigned int maxthreads, unsigned in
 }
 
 int QueueDevice::configure_tx(unsigned int maxthreads,unsigned int minqueues, unsigned int maxqueues, ErrorHandler *) {
-	usable_threads.assign(master()->nthreads(),true);
     _maxthreads = maxthreads;
     _minqueues = minqueues;
     _maxqueues = maxqueues;
@@ -67,7 +66,9 @@ int QueueDevice::configure_tx(unsigned int maxthreads,unsigned int minqueues, un
     return 0;
 }
 
-int QueueDevice::initialize_tx(ErrorHandler *) {
+int QueueDevice::initialize_tx(ErrorHandler * errh) {
+    usable_threads.assign(master()->nthreads(),false);
+    usable_threads = get_threads();
 
     int n_threads;
     if (_maxthreads == -1)
@@ -75,6 +76,10 @@ int QueueDevice::initialize_tx(ErrorHandler *) {
     else
         n_threads = min(_maxthreads,usable_threads.weight());
 
+    if (n_threads == 0) {
+        return errh->error("No threads end up in this queuedevice...? Aborting.");
+    }
+
     nqueues = min(_maxqueues,n_threads);
     nqueues = max(_minqueues,nqueues);
 

elements/userlevel/queuedevice.hh

@@ -139,6 +139,20 @@ protected:
         thread_state->_dropped += n;
     }
 
+    bool get_runnable_threads(Bitvector& bmk)
+    {
+    	if (noutputs()) { //RX
+    		for (int i = 0; i < nqueues; i++) {
+    			for (int j = 0; j < queue_share; j++) {
+    				bmk[thread_for_queue(i) - j] = 1;
+    			}
+    		}
+    		return true;
+    	} else { //TX
+    		return true;
+    	}
+    }
+
     /*
      * Configure a RX side of a queuedevice. Take cares of setting user max
      *  threads, queues and offset and registering this rx device for later

elements/userlevel/tonetmapdevice.cc

@@ -60,8 +60,6 @@ ToNetmapDevice::configure(Vector<String> &conf, ErrorHandler *errh)
     }
     configure_tx(maxthreads,1,_device->n_queues,errh); //Using the fewer possible number of queues is the better
 
-
-
     if (_burst > _device->some_nmd->some_ring->num_slots / 2) {
         errh->warning("BURST value larger than half the ring size (%d) is not recommended. Please set BURST to %d or less",_burst, _device->some_nmd->some_ring->num_slots,_device->some_nmd->some_ring->num_slots/2);
     }
@@ -73,18 +71,6 @@ int ToNetmapDevice::initialize(ErrorHandler *errh)
 {
     int ret;
 
-    //If full push, and a thread is assign, we set it as the only thread which can push packets to this element
-    if (!input_is_pull(0) && router()->thread_sched() != NULL && router()->thread_sched()->initial_home_thread_id(this) != ThreadSched::THREAD_UNKNOWN) {
-    	int thread_id = router()->thread_sched()->initial_home_thread_id(this);
-    	click_chatter("Element %s is assigned to thread %d : double check that only this thread will push packets to this element !\n", name().c_str(),thread_id); // Will be auto-detected by bitvector
-        usable_threads.assign(master()->nthreads(),false);
-        usable_threads[thread_id] = true;
-    } else {
-    	int thread_id = this->home_thread()->thread_id();
-    	usable_threads.assign(master()->nthreads(),false);
-    	usable_threads[thread_id] = true;
-    }
-
     ret = initialize_tx(errh);
     if (ret != 0)
         return ret;

include/click/element.hh

@@ -45,6 +45,10 @@ class Element { public:
     virtual void selected(int fd);
 #endif
 
+    virtual bool get_runnable_threads(Bitvector& b);
+    inline bool is_fullpush() const;
+    Bitvector get_threads();
+
     inline void checked_output_push(int port, Packet *p) const;
     inline Packet* checked_input_pull(int port) const;
 
@@ -268,6 +272,7 @@ class Element { public:
 
     Router* _router;
     int _eindex;
+    bool _is_fullpush;
 
 #if CLICK_STATS >= 2
     // STATISTICS
@@ -684,6 +689,19 @@ Element::Port::pull() const
     return p;
 }
 
+/**
+ * @brief Tell if the path up to this element is a full push path
+ *
+ * @pre get_threads() have to be called on this element or any downstream element
+ *
+ * If this element is part of a full push path, it means that packets passing
+ *  through will always be handled by the same thread and are not shared.
+ */
+inline bool Element::is_fullpush() const {
+    return _is_fullpush;
+}
+
+
 /** @brief Push packet @a p to output @a port, or kill it if @a port is out of
  * range.
  *

include/click/router.hh

@@ -90,6 +90,8 @@ class Router { public:
     inline ThreadSched* thread_sched() const;
     inline void set_thread_sched(ThreadSched* scheduler);
     inline int home_thread_id(const Element *e) const;
+    inline bool is_fullpush() const;
+    inline void non_fullpush();
 
     /** @cond never */
     // Needs to be public for NameInfo, but not useful outside
@@ -290,6 +292,7 @@ class Router { public:
     HashMap_ArenaFactory* _arena_factory;
     Router* _hotswap_router;
     ThreadSched* _thread_sched;
+    bool _is_fullpush;
     mutable NameInfo* _name_info;
     Vector<int> _flow_code_override_eindex;
     Vector<String> _flow_code_override;
@@ -452,6 +455,25 @@ Router::home_thread_id(const Element *e) const
 	return hard_home_thread_id(e);
 }
 
+/**
+ * @brief Tell if the whole router is in full push mode
+ *
+ * @pre getThreads() has been called on all final elements
+ *
+ * If true, all packets of this router are always handled by the same thread
+ *  (but mutltiple threads can handle different packets). This means that
+ *  sharing can be disabled.
+ */
+inline bool
+Router::is_fullpush() const {
+    return _is_fullpush;
+}
+
+inline void
+Router::non_fullpush() {
+    _is_fullpush = false;
+}
+
 /** @cond never */
 /** @brief  Return the NameInfo object for this router, if it exists.
  *

lib/element.cc

@@ -31,6 +31,7 @@
 #include <click/master.hh>
 #include <click/straccum.hh>
 #include <click/etheraddress.hh>
+#include <click/routervisitor.hh>
 #if CLICK_DEBUG_SCHEDULING
 # include <click/notifier.hh>
 #endif
@@ -1663,6 +1664,35 @@ Element::home_thread() const
     return master()->thread(router()->home_thread_id(this));
 }
 
+class InputThreadVisitor: public RouterVisitor {
+public:
+     Bitvector& bitmask;
+     bool fullpush;
+
+     InputThreadVisitor(Bitvector& b): bitmask(b), fullpush(true) {}
+
+     virtual bool visit(Element *e, bool isoutput, int port,
+                   Element *from_e, int from_port, int distance) {
+         bool ret = e->get_runnable_threads(bitmask);
+         if (ret == false)
+             fullpush = false;
+         return ret;
+     }
+};
+
+bool Element::get_runnable_threads(Bitvector& bmp) {
+    return true;
+}
+
+Bitvector Element::get_threads() {
+    Bitvector b(master()->nthreads());
+    InputThreadVisitor visitor(b);
+    router()->visit(this,false,-1,&visitor);
+    _is_fullpush = visitor.fullpush;
+    if (!_is_fullpush)
+        router()->non_fullpush();
+    return b;
+}
 
 // SELECT
 

lib/router.cc

@@ -83,7 +83,7 @@ Router::Router(const String &configuration, Master *master)
       _configuration(configuration),
       _notifier_signals(0),
       _arena_factory(new HashMap_ArenaFactory),
-      _hotswap_router(0), _thread_sched(0), _name_info(0), _next_router(0)
+      _hotswap_router(0), _thread_sched(0), _is_fullpush(true), _name_info(0), _next_router(0)
 {
     _refcount = 0;
     _runcount = 0;