addition of a fuzzing client and subsequent rewrite of parts of the network communication and SSL detection

Addition of a fuzzing client and subsequent rewrite of parts of the network communication and SSL detection in order to be more robust against horribly incorrect input from the client

Authored-by: Bruggeman, Otto G <[email protected]>

Author: rdementi

src/debug.h

@@ -32,9 +32,10 @@ namespace debug {
     template<typename LVL, typename PF, typename F, typename L, typename... Args>
     void dyn_debug_output( std::ostream& out, LVL level, PF pretty_function, F file, L line, Args... args ) {
         std::stringstream ss;
+        auto now = time(nullptr);
         ss << "DBG(" << std::dec << level << "): File '" << file << "', line '" << std::dec << line << "' :\n";
         ss << "DBG(" << std::dec << level << "): " << pretty_function << ":\n";
-        ss << "DBG(" << std::dec << level << "): "; // Next code line will continue printing on this output line
+        ss << "DBG(" << std::dec << level << ") " << std::put_time( localtime(&now), "%F_%T: " ); // Next code line will continue printing on this output line
         dyn_debug_output_helper( ss, args... );
         out << ss.str() << std::flush;
     }
@@ -57,6 +58,6 @@ namespace debug {
 
 #define DBG( level, ... ) \
     if ( debug::currentDebugLevel >= level ) \
-        debug::dyn_debug_output( std::cout, level, __PRETTY_FUNCTION__, __FILE__, __LINE__, __VA_ARGS__)
+        debug::dyn_debug_output( std::cerr, level, __PRETTY_FUNCTION__, __FILE__, __LINE__, __VA_ARGS__)
 
 } // namespace pcm

src/pcm-sensor-server.cpp

@@ -11,8 +11,12 @@ void ThreadPool::execute( ThreadPool* tp ) {
         Work* w = tp->retrieveWork();
         if ( w == nullptr ) break;
         w->execute();
-        deleteAndNullify(w);
+        // There can never be a double delete here, once taken from the tp it is owned by this thread
+        // but in order to silence cppcheck w is set explicitly to null
+        deleteAndNullify( w );
+        DBG( 5, "Work deleted, waiting for more work..." );
     }
+    DBG( 4, "Thread is explicitly dying now..." );
 }
 
 } // namespace pcm

src/threadpool.cpp

@@ -55,7 +55,7 @@ class ThreadPool {
     ThreadPool & operator = ( ThreadPool const& ) = delete;
 
 public:
-    ~ThreadPool() {
+    void emptyThreadPool( void ) {
         try {
             for (size_t i = 0; i < threads_.size(); ++i)
                 addWork(nullptr);
@@ -69,28 +69,33 @@ class ThreadPool {
         }
     }
 
+    ~ThreadPool() {
+        DBG( 5, "Threadpool is being deleted..." );
+        emptyThreadPool();
+    }
+
 public:
     static ThreadPool& getInstance() {
         static ThreadPool tp_(64);
         return tp_;
     }
 
     void addWork( Work* w ) {
-        DBG( 3, "WQ: Adding work" );
+        DBG( 5, "WQ: Adding work" );
         std::lock_guard<std::mutex> lg( qMutex_ );
         workQ_.push( w );
         queueCV_.notify_one();
-        DBG( 3, "WQ: Work available" );
+        DBG( 5, "WQ: Work available" );
     }
 
     Work* retrieveWork() {
-        DBG( 3, "WQ: Retrieving work" );
+        DBG( 5, "WQ: Retrieving work" );
         std::unique_lock<std::mutex> lock( qMutex_ );
         queueCV_.wait( lock, [this]{ return !workQ_.empty(); } );
         Work* w = workQ_.front();
         workQ_.pop();
         lock.unlock();
-        DBG( 3, "WQ: Work retrieved" );
+        DBG( 5, "WQ: Work retrieved" );
 
         return w;
     }
@@ -111,12 +116,23 @@ class ThreadPool {
 };
 
 class WorkQueue {
-public:
-    WorkQueue() : tp_( ThreadPool::getInstance() ), workProcessed_(0) {}
+private:
+    WorkQueue( size_t init ) : tp_( ThreadPool::getInstance() ), workProcessed_( init ) {
+        DBG( 5, "Constructing WorkQueue..." );
+    }
     WorkQueue( WorkQueue const& ) = delete;
     WorkQueue & operator = ( WorkQueue const& ) = delete;
-    ~WorkQueue() = default;
 
+public:
+    ~WorkQueue() {
+        DBG( 5, "Destructing WorkQueue..." );
+    }
+
+public:
+    static WorkQueue* getInstance() {
+        static WorkQueue wq_( 0 );
+        return &wq_;
+    }
     // Just forwarding to the threadpool
     void addWork( Work* w ) {
         ++workProcessed_;

src/threadpool.h

@@ -102,15 +102,4 @@ void Aggregator::dispatch( SystemRoot const& syp ) {
     readAccelCounters(sycs_);
 }
 
-Aggregator::Aggregator()
-{
-    PCM* const pcm = PCM::getInstance();
-    // Resize user provided vectors to the right size
-    ccsVector_.resize( pcm->getNumCores() );
-    socsVector_.resize( pcm->getNumSockets() );
-    // Internal use only, need to be the same size as the user provided vectors
-    ccsFutures_.resize( pcm->getNumCores() );
-    ucsFutures_.resize( pcm->getNumSockets() );
-}
-
 }// namespace pcm

src/topology.cpp

@@ -451,8 +451,20 @@ class SystemRoot : public SystemObject {
 class Aggregator : Visitor
 {
 public:
-    Aggregator();
-    virtual ~Aggregator() {}
+    Aggregator() : wq_( WorkQueue::getInstance() )
+    {
+        PCM* const pcm = PCM::getInstance();
+        // Resize user provided vectors to the right size
+        ccsVector_.resize( pcm->getNumCores() );
+        socsVector_.resize( pcm->getNumSockets() );
+        // Internal use only, need to be the same size as the user provided vectors
+        ccsFutures_.resize( pcm->getNumCores() );
+        ucsFutures_.resize( pcm->getNumSockets() );
+    }
+
+    virtual ~Aggregator() {
+        wq_ = nullptr;
+    }
 
 public:
     virtual void dispatch( SystemRoot const& syp ) override;
@@ -465,17 +477,15 @@ class Aggregator : Visitor
         // Fetch UncoreCounterState async result
         auto job = new LambdaJob<UncoreCounterState>(
             []( Socket* s ) -> UncoreCounterState {
-                DBG( 3, "Lambda fetching UncoreCounterState async" );
+                DBG( 5, "Lambda fetching UncoreCounterState async" );
                 UncoreCounterState ucs;
                 if ( !s->isOnline() )
                     return ucs;
                 return s->uncore()->uncoreCounterState();
             }, sop
         );
         ucsFutures_[ sop->socketID() ] = job->getFuture();
-        wq_.addWork( job );
-        // For now execute directly to compile test
-        //job->execute();
+        wq_->addWork( job );
     }
 
     virtual void dispatch( Core* cop ) override {
@@ -492,15 +502,15 @@ class Aggregator : Visitor
         // std::cerr << "Dispatch htp with osID=" << htp->osID() << "\n";
         auto job = new LambdaJob<CoreCounterState>(
             []( HyperThread* h ) -> CoreCounterState {
-                DBG( 3, "Lambda fetching CoreCounterState async" );
+                DBG( 5, "Lambda fetching CoreCounterState async" );
                 CoreCounterState ccs;
                 if ( !h->isOnline() )
                     return ccs;
                 return h->coreCounterState();
             }, htp
         );
         ccsFutures_[ htp->osID() ] = job->getFuture();
-        wq_.addWork( job );
+        wq_->addWork( job );
     }
 
     virtual void dispatch( ServerUncore* /*sup*/ ) override {
@@ -528,13 +538,13 @@ class Aggregator : Visitor
     }
 
 private:
+    WorkQueue* wq_;
     std::vector<CoreCounterState> ccsVector_;
     std::vector<SocketCounterState> socsVector_;
     SystemCounterState sycs_;
     std::vector<std::future<CoreCounterState>> ccsFutures_;
     std::vector<std::future<UncoreCounterState>> ucsFutures_;
     std::chrono::steady_clock::time_point dispatchedAt_{};
-    WorkQueue wq_;
 };
 
 } // namespace pcm