pstats: Proper thread cleanup after a thread exits (PStats 3.2)

The thread will linger around in the server as long as there is data (so, it will be removed after `pstats-history` seconds)

Fixes #450

panda/src/pipeline/thread.cxx

@@ -66,6 +66,10 @@ Thread::
   nassertv(_blocked_on_mutex == nullptr &&
            _waiting_on_cvar == nullptr);
 #endif
+
+  if (_pstats_callback != nullptr) {
+    _pstats_callback->delete_hook(this);
+  }
 }
 
 /**
@@ -251,3 +255,11 @@ deactivate_hook(Thread *) {
 void Thread::PStatsCallback::
 activate_hook(Thread *) {
 }
+
+/**
+ * Called when the thread is deleted.  This provides a callback hook for PStats
+ * to remove a thread's data when the thread is removed.
+ */
+void Thread::PStatsCallback::
+delete_hook(Thread *) {
+}

panda/src/pipeline/thread.h

@@ -128,6 +128,7 @@ public:
     virtual ~PStatsCallback();
     virtual void deactivate_hook(Thread *thread);
     virtual void activate_hook(Thread *thread);
+    virtual void delete_hook(Thread *thread);
   };
 
   INLINE void set_pstats_index(int pstats_index);

panda/src/pstatclient/pStatClient.cxx

@@ -27,6 +27,8 @@
 #include "clockObject.h"
 #include "neverFreeMemory.h"
 
+#include <algorithm>
+
 using std::string;
 
 PStatCollector PStatClient::_heap_total_size_pcollector("System memory:Heap");
@@ -415,6 +417,7 @@ client_main_tick() {
            vi != indices.end();
            ++vi) {
         InternalThread *thread = get_thread_ptr(*vi);
+        nassertd(thread != nullptr) continue;
         _impl->new_frame(*vi, thread->_frame_number);
         thread->_frame_number = clock->get_frame_count(get_thread_object(*vi));
       }
@@ -483,10 +486,12 @@ client_disconnect() {
   ThreadPointer *threads = _threads.load(std::memory_order_relaxed);
   for (int ti = 0; ti < get_num_threads(); ++ti) {
     InternalThread *thread = threads[ti];
-    thread->_frame_number = 0;
-    thread->_is_active = false;
-    thread->_next_packet = 0.0;
-    thread->_frame_data.clear();
+    if (thread != nullptr) {
+      thread->_frame_number = 0;
+      thread->_is_active = false;
+      thread->_next_packet = 0.0;
+      thread->_frame_data.clear();
+    }
   }
 
   CollectorPointer *collectors = _collectors.load(std::memory_order_relaxed);
@@ -1233,6 +1238,54 @@ activate_hook(Thread *thread) {
   }
 }
 
+/**
+ * Called when the thread is deleted.  This provides a callback hook for PStats
+ * to remove a thread's data when the thread is removed.
+ */
+void PStatClient::
+delete_hook(Thread *thread) {
+  int thread_index = thread->get_pstats_index();
+  if (thread_index < 0) {
+    return;
+  }
+
+  PStatClientImpl *impl;
+  InternalThread *ithread;
+  {
+    ReMutexHolder holder(_lock);
+    impl = _impl;
+    if (impl == nullptr) {
+      return;
+    }
+
+    if (!impl->client_is_connected()) {
+      return;
+    }
+
+    MultiThingsByName::iterator ni;
+
+    ni = _threads_by_name.find(thread->get_name());
+    if (ni != _threads_by_name.end()) {
+      ni->second.erase(std::remove(ni->second.begin(), ni->second.end(), thread_index));
+    }
+
+    ni = _threads_by_sync_name.find(thread->get_sync_name());
+    if (ni != _threads_by_sync_name.end()) {
+      ni->second.erase(std::remove(ni->second.begin(), ni->second.end(), thread_index));
+    }
+
+    // This load can be relaxed because we hold the lock.
+    ThreadPointer *threads = _threads.load(std::memory_order_relaxed);
+    ithread = threads[thread_index];
+    ithread->_is_active = false;
+    ithread->_thread_active = false;
+    threads[thread_index] = nullptr;
+  }
+
+  impl->remove_thread(thread_index);
+  delete ithread;
+}
+
 /**
  * Creates the new PStatCollectorDef for this collector.
  */

panda/src/pstatclient/pStatClient.h

@@ -151,6 +151,7 @@ private:
 
   virtual void deactivate_hook(Thread *thread);
   virtual void activate_hook(Thread *thread);
+  virtual void delete_hook(Thread *thread);
 
 private:
   // This mutex protects everything in this class.

panda/src/pstatclient/pStatClientImpl.cxx

@@ -328,6 +328,7 @@ new_frame(int thread_index, int frame_number) {
   nassertv(thread_index >= 0 && thread_index < _client->_num_threads);
 
   PStatClient::InternalThread *pthread = _client->get_thread_ptr(thread_index);
+  nassertv(pthread != nullptr);
 
   // If we're the main thread, we should exchange control packets with the
   // server.
@@ -428,6 +429,7 @@ add_frame(int thread_index, int frame_number, PStatFrameData &&frame_data) {
   nassertv(thread_index >= 0 && thread_index < _client->_num_threads);
 
   PStatClient::InternalThread *pthread = _client->get_thread_ptr(thread_index);
+  nassertv(pthread != nullptr);
 
   // If we're the main thread, we should exchange control packets with the
   // server.
@@ -458,6 +460,22 @@ add_frame(int thread_index, int frame_number, PStatFrameData &&frame_data) {
   _client->stop(pstats_index, current_thread_index);
 }
 
+/**
+ * Removes a thread from PStats.
+ */
+void PStatClientImpl::
+remove_thread(int thread_index) {
+  nassertv(thread_index >= 0 && thread_index < _client->_num_threads);
+
+  PStatClientControlMessage message;
+  message._type = PStatClientControlMessage::T_expire_thread;
+  message._first_thread_index = thread_index;
+
+  Datagram datagram;
+  message.encode(datagram);
+  _writer.send(datagram, _tcp_connection, true);
+}
+
 /**
  * Passes off the frame data to the writer thread.  If threading is disabled,
  * transmits it right away.
@@ -541,6 +559,8 @@ transmit_frame_data(int thread_index, int frame_number,
                     const PStatFrameData &frame_data) {
   nassertv(thread_index >= 0 && thread_index < _client->_num_threads);
   PStatClient::InternalThread *thread = _client->get_thread_ptr(thread_index);
+  nassertv(thread != nullptr);
+
   if (_is_connected && thread->_is_active) {
 
     // We don't want to send too many packets in a hurry and flood the server.
@@ -680,12 +700,6 @@ send_hello() {
   message._major_version = get_current_pstat_major_version();
   message._minor_version = get_current_pstat_minor_version();
 
-  // The Python profiling feature may send nested start/stop pairs, so requires
-  // a server version capable of dealing with this.
-  if (pstats_python_profiler && message._major_version <= 3) {
-    message._major_version = 3;
-    message._minor_version = std::max(message._minor_version, 1);
-  }
 
   Datagram datagram;
   message.encode(datagram);
@@ -733,7 +747,11 @@ report_new_threads() {
     PStatClient::ThreadPointer *threads =
       (PStatClient::ThreadPointer *)_client->_threads;
     while (_threads_reported < _client->_num_threads) {
-      message._names.push_back(threads[_threads_reported]->_name);
+      if (threads[_threads_reported] != nullptr) {
+        message._names.push_back(threads[_threads_reported]->_name);
+      } else {
+        message._names.push_back(std::string());
+      }
       _threads_reported++;
     }
 

panda/src/pstatclient/pStatClientImpl.h

@@ -70,6 +70,8 @@ public:
   void new_frame(int thread_index, int frame_number = -1);
   void add_frame(int thread_index, int frame_number, PStatFrameData &&frame_data);
 
+  void remove_thread(int thread_index);
+
 private:
   void enqueue_frame_data(int thread_index, int frame_number,
                           PStatFrameData &&frame_data);

panda/src/pstatclient/pStatFrameData.cxx

@@ -44,7 +44,7 @@ write_datagram(Datagram &destination, PStatClient *client) const {
 
 #if !defined(WORDS_BIGENDIAN) || defined(__GNUC__)
   // Hand-roll this, significantly more efficient for many data points
-  size_t size = (_time_data.size() + _level_data.size()) * 6 + 4;
+  size_t size = (_time_data.size() + _level_data.size()) * 6 + 8;
   PTA_uchar array = destination.modify_array();
   size_t offset = array.size();
   array.resize(offset + size);
@@ -53,7 +53,8 @@ write_datagram(Datagram &destination, PStatClient *client) const {
   uint16_t *ptr = (uint16_t *)data;
 
 #ifdef WORDS_BIGENDIAN
-  *ptr++ = __builtin_bswap16(_time_data.size());
+  *(uint32_t *)ptr = __builtin_bswap32(_time_data.size());
+  ptr += 2;
 
   for (const DataPoint &dp : _time_data) {
     *ptr++ = __builtin_bswap16(dp._index);
@@ -62,7 +63,9 @@ write_datagram(Datagram &destination, PStatClient *client) const {
     ptr += 2;
   }
 
-  *ptr++ = __builtin_bswap16(_level_data.size());
+  *(uint32_t *)ptr = __builtin_bswap16(_level_data.size());
+  ptr += 2;
+
   for (const DataPoint &dp : _level_data) {
     *ptr++ = __builtin_bswap16(dp._index);
     PN_float32 v = (PN_float32)dp._value;
@@ -70,7 +73,8 @@ write_datagram(Datagram &destination, PStatClient *client) const {
     ptr += 2;
   }
 #else
-  *ptr++ = _time_data.size();
+  *(uint32_t *)ptr = _time_data.size();
+  ptr += 2;
 
   for (const DataPoint &dp : _time_data) {
     *ptr++ = dp._index;
@@ -78,7 +82,9 @@ write_datagram(Datagram &destination, PStatClient *client) const {
     ptr += 2;
   }
 
-  *ptr++ = _level_data.size();
+  *(uint32_t *)ptr = _level_data.size();
+  ptr += 2;
+
   for (const DataPoint &dp : _level_data) {
     *ptr++ = dp._index;
     *(PN_float32 *)ptr = dp._value;
@@ -87,12 +93,12 @@ write_datagram(Datagram &destination, PStatClient *client) const {
 #endif
 
 #else
-  destination.add_uint16(_time_data.size());
+  destination.add_uint32(_time_data.size());
   for (const DataPoint &dp : _time_data) {
     destination.add_uint16(dp._index);
     destination.add_float32(dp._value);
   }
-  destination.add_uint16(_level_data.size());
+  destination.add_uint32(_level_data.size());
   for (const DataPoint &dp : _level_data) {
     destination.add_uint16(dp._index);
     destination.add_float32(dp._value);

panda/src/pstatclient/pStatProperties.cxx

@@ -26,7 +26,7 @@
 using std::string;
 
 static const int current_pstat_major_version = 3;
-static const int current_pstat_minor_version = 0;
+static const int current_pstat_minor_version = 2;
 // Initialized at 2.0 on 5/18/01, when version numbers were first added.
 // Incremented to 2.1 on 5/21/01 to add support for TCP frame data.
 // Incremented to 3.0 on 4/28/05 to bump TCP headers to 32 bits.

pandatool/src/gtk-stats/gtkStatsChartMenu.h

@@ -32,6 +32,8 @@ public:
   GtkStatsChartMenu(GtkStatsMonitor *monitor, int thread_index);
   ~GtkStatsChartMenu();
 
+  int get_thread_index() const { return _thread_index; }
+
   GtkWidget *get_menu_widget();
   void add_to_menu_bar(GtkWidget *menu_bar, int position);
   void remove_from_menu_bar(GtkWidget *menu_bar);

pandatool/src/gtk-stats/gtkStatsMonitor.cxx

@@ -207,6 +207,23 @@ new_data(int thread_index, int frame_number) {
   }
 }
 
+/**
+ * Called when a thread should be removed from the list of threads.
+ */
+void GtkStatsMonitor::
+remove_thread(int thread_index) {
+  for (ChartMenus::iterator it = _chart_menus.begin(); it != _chart_menus.end(); ++it) {
+    GtkStatsChartMenu *chart_menu = *it;
+    if (chart_menu->get_thread_index() == thread_index) {
+      chart_menu->remove_from_menu_bar(_menu_bar);
+      delete chart_menu;
+      _chart_menus.erase(it);
+      --_next_chart_index;
+      return;
+    }
+  }
+}
+
 /**
  * Called whenever the connection to the client has been lost.  This is a
  * permanent state change.  The monitor should update its display to represent

pandatool/src/gtk-stats/gtkStatsMonitor.h

@@ -71,6 +71,7 @@ public:
   virtual void new_collector(int collector_index);
   virtual void new_thread(int thread_index);
   virtual void new_data(int thread_index, int frame_number);
+  virtual void remove_thread(int thread_index);
   virtual void lost_connection();
   virtual void idle();
   virtual bool has_idle();

pandatool/src/gtk-stats/gtkStatsTimeline.cxx

@@ -283,6 +283,9 @@ end_draw() {
 
     int max_width = 0;
     for (const ThreadRow &thread_row : _threads) {
+      if (!thread_row._visible) {
+        continue;
+      }
       pango_layout_set_text(layout, thread_row._label.c_str(), thread_row._label.size());
 
       int width, height;
@@ -786,7 +789,9 @@ draw_guide_label(cairo_t *cr, const PStatGraph::GuideBar &bar) {
 void GtkStatsTimeline::
 draw_thread_labels(cairo_t *cr) {
   for (const ThreadRow &thread_row : _threads) {
-    draw_thread_label(cr, thread_row);
+    if (thread_row._visible) {
+      draw_thread_label(cr, thread_row);
+    }
   }
 }
 

pandatool/src/pstatserver/pStatClientData.cxx

@@ -80,6 +80,22 @@ close() {
   }
 }
 
+/**
+ * Returns the timestamp (in seconds elapsed since connection) of the latest
+ * available frame.
+ */
+double PStatClientData::
+get_latest_time() const {
+  double time = 0.0;
+  for (const Thread &thread : _threads) {
+    if (thread._data != nullptr && !thread._data->is_empty()) {
+      time = std::max(time, thread._data->get_latest_time());
+    }
+  }
+
+  return time;
+}
+
 /**
  * Returns the total number of collectors the Data knows about.
  */
@@ -244,7 +260,7 @@ get_num_threads() const {
  */
 bool PStatClientData::
 has_thread(int index) const {
-  return (index >= 0 && index < (int)_threads.size() &&
+  return (index >= 0 && (size_t)index < _threads.size() &&
           !_threads[index]._name.empty());
 }
 
@@ -281,10 +297,18 @@ get_thread_name(int index) const {
 const PStatThreadData *PStatClientData::
 get_thread_data(int index) const {
   ((PStatClientData *)this)->define_thread(index);
-  nassertr(index >= 0 && index < (int)_threads.size(), nullptr);
+  nassertr(index >= 0 && (size_t)index < _threads.size(), nullptr);
   return _threads[index]._data;
 }
 
+/**
+ * Returns true if the given thread is still alive.
+ */
+bool PStatClientData::
+is_thread_alive(int index) const {
+  return (index >= 0 && (size_t)index < _threads.size() && _threads[index]._is_alive);
+}
+
 /**
  * Returns the number of Collectors between the indicated parent and the child
  * Collector in the relationship graph.  If child is the same as parent,
@@ -346,7 +370,7 @@ add_collector(PStatCollectorDef *def) {
  * information just arrived from the client.
  */
 void PStatClientData::
-define_thread(int thread_index, const string &name) {
+define_thread(int thread_index, const string &name, bool mark_alive) {
   // A sanity check on the index number.
   nassertv(thread_index < 1000);
 
@@ -355,6 +379,10 @@ define_thread(int thread_index, const string &name) {
     _threads.push_back(Thread());
   }
 
+  if (mark_alive) {
+    _threads[thread_index]._is_alive = true;
+  }
+
   if (!name.empty()) {
     _threads[thread_index]._name = name;
   }
@@ -366,6 +394,29 @@ define_thread(int thread_index, const string &name) {
   _is_dirty = true;
 }
 
+/**
+ * Indicates that the given thread has expired.  Presumably this is information
+ * just arrived from the client.
+ */
+void PStatClientData::
+expire_thread(int thread_index) {
+  if (thread_index >= 0 && (size_t)thread_index < _threads.size()) {
+    _threads[thread_index]._is_alive = false;
+  }
+}
+
+/**
+ * Removes the given thread data entirely.
+ */
+void PStatClientData::
+remove_thread(int thread_index) {
+  if (thread_index >= 0 && (size_t)thread_index < _threads.size()) {
+    _threads[thread_index]._name.clear();
+    _threads[thread_index]._data.clear();
+    _threads[thread_index]._is_alive = false;
+  }
+}
+
 /**
  * Makes room for and stores a new frame's worth of data associated with some
  * particular thread (which may or may not have already been defined).
@@ -469,7 +520,7 @@ read_datagram(DatagramIterator &scan) {
   int thread_index;
   while ((thread_index = scan.get_int16()) != -1) {
     std::string name = scan.get_string();
-    define_thread(thread_index, name);
+    define_thread(thread_index, name, true);
 
     _threads[thread_index]._data->read_datagram(scan, this);
   }

pandatool/src/pstatserver/pStatClientData.h

@@ -45,6 +45,8 @@ public:
   bool is_alive() const;
   void close();
 
+  double get_latest_time() const;
+
   int get_num_collectors() const;
   bool has_collector(int index) const;
   int find_collector(const std::string &fullname) const;
@@ -62,12 +64,16 @@ public:
   int find_thread(const std::string &name) const;
   std::string get_thread_name(int index) const;
   const PStatThreadData *get_thread_data(int index) const;
+  bool is_thread_alive(int index) const;
 
   int get_child_distance(int parent, int child) const;
 
 
   void add_collector(PStatCollectorDef *def);
-  void define_thread(int thread_index, const std::string &name = std::string());
+  void define_thread(int thread_index, const std::string &name = std::string(),
+                     bool mark_alive = false);
+  void expire_thread(int thread_index);
+  void remove_thread(int thread_index);
 
   void record_new_frame(int thread_index, int frame_number,
                         PStatFrameData *frame_data);
@@ -101,6 +107,7 @@ private:
   public:
     std::string _name;
     PT(PStatThreadData) _data;
+    bool _is_alive = false;
   };
   typedef pvector<Thread> Threads;
   Threads _threads;

pandatool/src/pstatserver/pStatMonitor.cxx

@@ -566,6 +566,13 @@ void PStatMonitor::
 new_data(int, int) {
 }
 
+/**
+ * Called when a thread should be removed from the list of threads.
+ */
+void PStatMonitor::
+remove_thread(int) {
+}
+
 /**
  * Called whenever the connection to the client has been lost.  This is a
  * permanent state change.  The monitor should update its display to represent

pandatool/src/pstatserver/pStatMonitor.h

@@ -92,6 +92,7 @@ public:
   virtual void new_collector(int collector_index);
   virtual void new_thread(int thread_index);
   virtual void new_data(int thread_index, int frame_number);
+  virtual void remove_thread(int thread_index);
 
   virtual void lost_connection();
   virtual void idle();

pandatool/src/pstatserver/pStatReader.cxx

@@ -194,8 +194,7 @@ handle_client_control_message(const PStatClientControlMessage &message) {
 
       if (message._major_version != server_major_version ||
           (message._major_version == server_major_version &&
-           message._minor_version > server_minor_version &&
-           (message._major_version != 3 || message._minor_version > 2))) {
+           message._minor_version > server_minor_version)) {
         _monitor->bad_version(message._client_hostname, message._client_progname,
                               message._client_pid,
                               message._major_version, message._minor_version,
@@ -219,17 +218,39 @@ handle_client_control_message(const PStatClientControlMessage &message) {
 
   case PStatClientControlMessage::T_define_threads:
     {
+      // See if we can clean up old threads, so that we don't clutter up the
+      // view if we are creating many threads.
+      for (int thread_index = 0; thread_index < _client_data->get_num_threads(); ++thread_index) {
+        if (_client_data->has_thread(thread_index) && !_client_data->is_thread_alive(thread_index)) {
+          PStatThreadData *thread_data = (PStatThreadData *)_client_data->get_thread_data(thread_index);
+          if (thread_data->prune_history(_client_data->get_latest_time())) {
+            _client_data->remove_thread(thread_index);
+            _monitor->remove_thread(thread_index);
+          }
+        }
+      }
+
       for (int i = 0; i < (int)message._names.size(); i++) {
         int thread_index = message._first_thread_index + i;
         std::string name = message._names[i];
-        _client_data->define_thread(thread_index, name);
+        _client_data->define_thread(thread_index, name, true);
         _monitor->new_thread(thread_index);
       }
     }
     break;
 
   case PStatClientControlMessage::T_expire_thread:
-    // Ignore for now.
+    if (_client_data->has_thread(message._first_thread_index)) {
+      // Remove the thread right away if it has no recent data.
+      PStatThreadData *thread_data = (PStatThreadData *)_client_data->get_thread_data(message._first_thread_index);
+      if (thread_data->prune_history(_client_data->get_latest_time())) {
+        _client_data->remove_thread(message._first_thread_index);
+        _monitor->remove_thread(message._first_thread_index);
+      } else {
+        // Otherwise, just mark it as expired, and we'll remove it later.
+        _client_data->expire_thread(message._first_thread_index);
+      }
+    }
     break;
 
   default:
@@ -278,7 +299,11 @@ handle_client_udp_data(const Datagram &datagram) {
  */
 void PStatReader::
 dequeue_frame_data() {
-  while (!_queued_frame_data.empty()) {
+  if (_queued_frame_data.empty()) {
+    return;
+  }
+
+  do {
     const FrameData &data = _queued_frame_data.front();
     nassertv(_client_data != nullptr);
 
@@ -300,4 +325,16 @@ dequeue_frame_data() {
 
     _queued_frame_data.pop_front();
   }
+  while (!_queued_frame_data.empty());
+
+  // Clean up old threads.
+  for (int thread_index = 0; thread_index < _client_data->get_num_threads(); ++thread_index) {
+    if (_client_data->has_thread(thread_index) && !_client_data->is_thread_alive(thread_index)) {
+      PStatThreadData *thread_data = (PStatThreadData *)_client_data->get_thread_data(thread_index);
+      if (thread_data->prune_history(_client_data->get_latest_time())) {
+        _client_data->remove_thread(thread_index);
+        _monitor->remove_thread(thread_index);
+      }
+    }
+  }
 }

pandatool/src/pstatserver/pStatThreadData.I

@@ -18,3 +18,11 @@ INLINE const PStatClientData *PStatThreadData::
 get_client_data() const {
   return _client_data;
 }
+
+/**
+ * Returns true if the structure contains no frames, false otherwise.
+ */
+INLINE bool PStatThreadData::
+is_empty() const {
+  return _frames.empty();
+}

pandatool/src/pstatserver/pStatThreadData.cxx

@@ -39,15 +39,6 @@ PStatThreadData::
 ~PStatThreadData() {
 }
 
-
-/**
- * Returns true if the structure contains no frames, false otherwise.
- */
-bool PStatThreadData::
-is_empty() const {
-  return _frames.empty();
-}
-
 /**
  * Returns the frame number of the most recent frame stored in the data.
  */
@@ -248,6 +239,25 @@ get_history() const {
   return _history;
 }
 
+/**
+ * Given a timestamp representing the time of the latest known frame, removes
+ * any frames older than the configured history.  Returns true if the data is
+ * now empty.
+ */
+bool PStatThreadData::
+prune_history(double time) {
+  double oldest_allowable_time = time - _history;
+  while (!_frames.empty() &&
+         (_frames.front() == nullptr ||
+          _frames.front()->is_time_empty() ||
+          _frames.front()->get_start() < oldest_allowable_time)) {
+    delete _frames.front();
+    _frames.pop_front();
+    _first_frame_number++;
+  }
+
+  return _frames.empty();
+}
 
 /**
  * Makes room for and stores a new frame's worth of data.  Calling this
@@ -261,27 +271,16 @@ void PStatThreadData::
 record_new_frame(int frame_number, PStatFrameData *frame_data) {
   nassertv(frame_data != nullptr);
   nassertv(!frame_data->is_empty());
-  double time = frame_data->get_start();
 
   // First, remove all the old frames that fall outside of our history window.
-  double oldest_allowable_time = time - _history;
-  while (!_frames.empty() &&
-         (_frames.front() == nullptr ||
-          _frames.front()->is_time_empty() ||
-          _frames.front()->get_start() < oldest_allowable_time)) {
-    delete _frames.front();
-    _frames.pop_front();
-    _first_frame_number++;
-  }
-
-  // Now, add enough empty frame definitions to account for the latest frame
+  // Then, add enough empty frame definitions to account for the latest frame
   // number.  This might involve some skips, since we don't guarantee that we
   // get all the frames in order or even at all.
-  if (_frames.empty()) {
+  if (prune_history(frame_data->get_start())) {
     _first_frame_number = frame_number;
     _frames.push_back(nullptr);
-
-  } else {
+  }
+  else {
     while (_first_frame_number + (int)_frames.size() <= frame_number) {
       _frames.push_back(nullptr);
     }

pandatool/src/pstatserver/pStatThreadData.h

@@ -40,7 +40,7 @@ public:
 
   INLINE const PStatClientData *get_client_data() const;
 
-  bool is_empty() const;
+  INLINE bool is_empty() const;
 
   int get_latest_frame_number() const;
   int get_oldest_frame_number() const;
@@ -60,6 +60,7 @@ public:
 
   void set_history(double time);
   double get_history() const;
+  bool prune_history(double time);
 
   void record_new_frame(int frame_number, PStatFrameData *frame_data);
 

pandatool/src/pstatserver/pStatTimeline.cxx

@@ -47,6 +47,11 @@ PStatTimeline(PStatMonitor *monitor, int xsize, int ysize) :
       ThreadRow &thread_row = _threads.back();
       thread_row._row_offset = row_offset;
 
+      if (!client_data->has_thread(thread_index)) {
+        continue;
+      }
+      thread_row._visible = true;
+
       const PStatThreadData *thread_data = client_data->get_thread_data(thread_index);
       if (thread_data != nullptr) {
         _threads_changed = true;
@@ -135,8 +140,11 @@ new_data(int thread_index, int frame_number) {
         } else {
           _threads.resize(_threads.size() + 1);
           _threads[_threads.size() - 1]._row_offset =
-            _threads[_threads.size() - 2]._row_offset +
-            _threads[_threads.size() - 2]._rows.size() + 1;
+            _threads[_threads.size() - 2]._row_offset;
+          if (_threads[_threads.size() - 2]._visible) {
+            _threads[_threads.size() - 1]._row_offset +=
+              _threads[_threads.size() - 2]._rows.size() + 1;
+          }
         }
       }
 
@@ -147,7 +155,9 @@ new_data(int thread_index, int frame_number) {
         size_t offset = thread_row._row_offset + thread_row._rows.size() + 1;
         for (size_t ti = (size_t)(thread_index + 1); ti < _threads.size(); ++ti) {
           _threads[ti]._row_offset = offset;
-          offset += _threads[ti]._rows.size() + 1;
+          if (_threads[ti]._visible) {
+            offset += _threads[ti]._rows.size() + 1;
+          }
         }
         _threads_changed = true;
         normal_guide_bars();
@@ -179,6 +189,11 @@ update_bars(int thread_index, int frame_number) {
   thread_row._label = client_data->get_thread_name(thread_index);
   bool changed_num_rows = false;
 
+  if (!thread_row._visible) {
+    thread_row._visible = true;
+    changed_num_rows = true;
+  }
+
   // pair<int collector_index, double start_time>
   pvector<std::pair<int, double> > stack;
 
@@ -480,11 +495,13 @@ force_redraw() {
 
   for (size_t ti = 0; ti < _threads.size(); ++ti) {
     ThreadRow &thread_row = _threads[ti];
-    for (size_t ri = 0; ri < thread_row._rows.size(); ++ri) {
-      draw_row((int)ti, (int)ri, start_time, end_time);
-      ++num_rows;
+    if (thread_row._visible) {
+      for (size_t ri = 0; ri < thread_row._rows.size(); ++ri) {
+        draw_row((int)ti, (int)ri, start_time, end_time);
+        ++num_rows;
+      }
+      draw_separator(num_rows++);
     }
-    draw_separator(num_rows++);
   }
 
   end_draw();
@@ -503,7 +520,7 @@ force_redraw(int row, int from_x, int to_x) {
 
   for (size_t ti = 0; ti < _threads.size(); ++ti) {
     ThreadRow &thread_row = _threads[ti];
-    if ((int)thread_row._row_offset > row) {
+    if (!thread_row._visible || (int)thread_row._row_offset > row) {
       break;
     }
 
@@ -654,8 +671,10 @@ draw_thread(int thread_index, double start_time, double end_time) {
   }
 
   ThreadRow &thread_row = _threads[(size_t)thread_index];
-  for (size_t ri = 0; ri < thread_row._rows.size(); ++ri) {
-    draw_row(thread_index, (int)ri, start_time, end_time);
+  if (thread_row._visible) {
+    for (size_t ri = 0; ri < thread_row._rows.size(); ++ri) {
+      draw_row(thread_index, (int)ri, start_time, end_time);
+    }
   }
 }
 

pandatool/src/pstatserver/pStatTimeline.h

@@ -102,6 +102,7 @@ protected:
     Rows _rows;
     size_t _row_offset = 0;
     int _last_frame = -1;
+    bool _visible = false;
   };
   typedef pvector<ThreadRow> ThreadRows;
   ThreadRows _threads;

pandatool/src/win-stats/winStatsChartMenu.cxx

@@ -61,13 +61,22 @@ add_to_menu_bar(HMENU menu_bar, int before_menu_id) {
   memset(&mii, 0, sizeof(mii));
   mii.cbSize = sizeof(mii);
 
-  mii.fMask = MIIM_STRING | MIIM_FTYPE | MIIM_SUBMENU;
+  mii.fMask = MIIM_STRING | MIIM_FTYPE | MIIM_SUBMENU | MIIM_ID;
   mii.fType = MFT_STRING;
+  mii.wID = 1000 | _thread_index;
   mii.hSubMenu = _menu;
   mii.dwTypeData = (char *)thread_name.c_str();
   InsertMenuItem(menu_bar, before_menu_id, FALSE, &mii);
 }
 
+/**
+ *
+ */
+void WinStatsChartMenu::
+remove_from_menu_bar(HMENU menu_bar) {
+  RemoveMenu(menu_bar, 1000 | _thread_index, MF_BYCOMMAND);
+}
+
 /**
  * Checks to see if the menu needs to be updated (e.g.  because of new data
  * from the client), and updates it if necessary.

pandatool/src/win-stats/winStatsChartMenu.h

@@ -33,8 +33,11 @@ public:
   WinStatsChartMenu(WinStatsMonitor *monitor, int thread_index);
   ~WinStatsChartMenu();
 
+  int get_thread_index() const { return _thread_index; }
+
   HMENU get_menu_handle();
   void add_to_menu_bar(HMENU menu_bar, int before_menu_id);
+  void remove_from_menu_bar(HMENU menu_bar);
 
   void check_update();
   void do_update();

pandatool/src/win-stats/winStatsMonitor.cxx

@@ -172,6 +172,22 @@ new_thread(int thread_index) {
   }
 }
 
+/**
+ * Called when a thread should be removed from the list of threads.
+ */
+void WinStatsMonitor::
+remove_thread(int thread_index) {
+  for (ChartMenus::iterator it = _chart_menus.begin(); it != _chart_menus.end(); ++it) {
+    WinStatsChartMenu *chart_menu = *it;
+    if (chart_menu->get_thread_index() == thread_index) {
+      chart_menu->remove_from_menu_bar(_menu_bar);
+      delete chart_menu;
+      _chart_menus.erase(it);
+      return;
+    }
+  }
+}
+
 /**
  * Called as each frame's data is made available.  There is no guarantee the
  * frames will arrive in order, or that all of them will arrive at all.  The

pandatool/src/win-stats/winStatsMonitor.h

@@ -70,6 +70,7 @@ public:
   virtual void new_collector(int collector_index);
   virtual void new_thread(int thread_index);
   virtual void new_data(int thread_index, int frame_number);
+  virtual void remove_thread(int thread_index);
   virtual void lost_connection();
   virtual void idle();
   virtual bool has_idle();

pandatool/src/win-stats/winStatsTimeline.cxx

@@ -620,7 +620,9 @@ additional_window_paint(HDC hdc) {
   SetTextAlign(hdc, TA_LEFT | TA_TOP | TA_NOUPDATECP);
 
   for (const ThreadRow &thread_row : _threads) {
-    draw_thread_label(hdc, thread_row);
+    if (thread_row._visible) {
+      draw_thread_label(hdc, thread_row);
+    }
   }
 }