Rwjs/unique sids

src/Main/Main.cc

@@ -98,6 +98,7 @@ int main(int argc, char* argv[]) {
 
         // Session manager
         session_manager = make_shared<SessionManager>(settings, auth_token, file_list_handler);
+        carta::OnMessageTask::SetSessionManager(session_manager);
 
         // HTTP server
         if (!settings.no_frontend || !settings.no_database || settings.enable_scripting) {

src/Session/OnMessageTask.cc

@@ -11,6 +11,8 @@
 
 using namespace carta;
 
+std::shared_ptr<SessionManager> OnMessageTask::_session_manager;
+
 OnMessageTask* SetImageChannelsTask::execute() {
     std::pair<CARTA::SetImageChannels, uint32_t> request_pair;
     bool tester;

src/Session/OnMessageTask.h

@@ -17,11 +17,15 @@
 
 #include "AnimationObject.h"
 #include "Session.h"
+#include "SessionManager.h"
 #include "Util/Message.h"
 
 namespace carta {
 
 class OnMessageTask {
+private:
+    static std::shared_ptr<SessionManager> _session_manager;
+
 protected:
     Session* _session;
 
@@ -31,10 +35,17 @@ class OnMessageTask {
     }
     virtual ~OnMessageTask() {
         if (!_session->DecreaseRefCount()) {
-            delete _session;
+            spdlog::info("({}) Remove Session {} in ~OMT", fmt::ptr(_session), _session->GetId());
+            // Test here since the CARTA test system does not set this shared_ptr for all tests.
+            if (_session_manager) {
+                _session_manager->DeleteSession(_session->GetId());
+            }
         }
         _session = nullptr;
     }
+    static void SetSessionManager(shared_ptr<SessionManager>& session_manager) {
+        _session_manager = session_manager;
+    }
     virtual OnMessageTask* execute() = 0;
 };
 

src/Session/Session.cc

@@ -92,7 +92,7 @@ void LoaderCache::Remove(const std::string& filename) {
     _queue.remove(filename);
 }
 
-int Session::_num_sessions = 0;
+volatile int Session::_num_sessions = 0;
 int Session::_exit_after_num_seconds = 5;
 bool Session::_exit_when_all_sessions_closed = false;
 std::thread* Session::_animation_thread = nullptr;
@@ -121,7 +121,7 @@ Session::Session(uWS::WebSocket<false, true, PerSocketData>* ws, uWS::Loop* loop
     _connected = true;
     ++_num_sessions;
     UpdateLastMessageTimestamp();
-    spdlog::debug("{} ::Session ({})", fmt::ptr(this), _num_sessions);
+    spdlog::info("{} ::Session ({}:{})", fmt::ptr(this), _id, _num_sessions);
 }
 
 static int __exit_backend_timer = 0;
@@ -147,17 +147,17 @@ void ExitNoSessions(int s) {
 
 Session::~Session() {
     --_num_sessions;
-    spdlog::debug("{} ~Session {}", fmt::ptr(this), _num_sessions);
+    spdlog::debug("{} ~Session : num sessions = {}", fmt::ptr(this), _num_sessions);
     if (!_num_sessions) {
         spdlog::info("No remaining sessions.");
         if (_exit_when_all_sessions_closed) {
             if (_exit_after_num_seconds == 0) {
-                spdlog::info("Exiting due to no sessions remaining");
-                ThreadManager::ExitEventHandlingThreads();
+                spdlog::debug("Exiting due to no sessions remaining");
                 logger::FlushLogFile();
-                exit(0);
+                __exit_backend_timer = 1;
+            } else {
+                __exit_backend_timer = _exit_after_num_seconds;
             }
-            __exit_backend_timer = _exit_after_num_seconds;
             struct sigaction sig_handler;
             sig_handler.sa_handler = ExitNoSessions;
             sigemptyset(&sig_handler.sa_mask);
@@ -363,6 +363,7 @@ void Session::OnRegisterViewer(const CARTA::RegisterViewer& message, uint16_t ic
     } else {
         type = CARTA::SessionType::RESUMED;
         if (session_id != _id) {
+            spdlog::info("({}) Session setting id to {} (was {}) on resume", fmt::ptr(this), session_id, _id);
             _id = session_id;
             status = fmt::format("Start a new backend and assign it with session id {}", session_id);
         } else {
@@ -1083,7 +1084,7 @@ void Session::OnSetContourParameters(const CARTA::SetContourParameters& message,
 
 void Session::OnResumeSession(const CARTA::ResumeSession& message, uint32_t request_id) {
     bool success(true);
-    spdlog::info("Client {} [{}] Resumed.", GetId(), GetAddress());
+    spdlog::info("Session {} [{}] Resumed.", GetId(), GetAddress());
 
     // Error messages
     std::string err_message;

src/Session/Session.h

@@ -288,9 +288,6 @@ class Session {
     void SendFileEvent(
         int file_id, CARTA::EventType event_type, u_int32_t event_id, google::protobuf::MessageLite& message, bool compress = true);
     void SendLogEvent(const std::string& message, std::vector<std::string> tags, CARTA::ErrorSeverity severity);
-    void StartAnimationThread() {
-        // Not sure if needed... XXX
-    }
 
     // uWebSockets
     uWS::WebSocket<false, true, PerSocketData>* _socket;
@@ -346,7 +343,7 @@ class Session {
     std::atomic<int> _ref_count;
     int _animation_id;
     bool _connected;
-    static int _num_sessions;
+    static volatile int _num_sessions;
     static int _exit_after_num_seconds;
     static bool _exit_when_all_sessions_closed;
     static std::thread* _animation_thread;

src/Session/SessionManager.cc

@@ -4,11 +4,12 @@
    SPDX-License-Identifier: GPL-3.0-or-later
 */
 
-#include "SessionManager.h"
-#include "ThreadingManager/ThreadingManager.h"
+#include <sys/time.h>
 
 #include "Logger/Logger.h"
 #include "OnMessageTask.h"
+#include "SessionManager.h"
+#include "ThreadingManager/ThreadingManager.h"
 #include "Util/Message.h"
 #include "Util/Token.h"
 
@@ -17,18 +18,24 @@ namespace carta {
 SessionManager::SessionManager(ProgramSettings& settings, std::string auth_token, std::shared_ptr<FileListHandler> file_list_handler)
     : _session_number(0), _app(uWS::App()), _settings(settings), _auth_token(auth_token), _file_list_handler(file_list_handler) {}
 
-void SessionManager::DeleteSession(int session_id) {
+void SessionManager::DeleteSession(uint32_t session_id) {
     Session* session = _sessions[session_id];
     if (session) {
         spdlog::info(
-            "Client {} [{}] Deleted. Remaining sessions: {}", session->GetId(), session->GetAddress(), Session::NumberOfSessions());
+            "Session {} [{}] Deleted. Remaining sessions: {}", session->GetId(), session->GetAddress(), Session::NumberOfSessions());
         session->WaitForTaskCancellation();
         session->CloseAllScriptingRequests();
-        if (!session->DecreaseRefCount()) {
+
+        if (!session->GetRefCount()) {
+            spdlog::info("Sessions in Session Map :");
+            for (const std::pair<uint32_t, Session*>& ssp : _sessions) {
+                Session* ss = ssp.second;
+                spdlog::info("\tMap id {}, session id {}, session ptr {}", ssp.first, ss->GetId(), fmt::ptr(ss));
+            }
             delete session;
             _sessions.erase(session_id);
         } else {
-            spdlog::warn("Session {} reference count is not 0 ({}) on deletion!", session_id, session->GetRefCount());
+            spdlog::info("Session {} reference count is not 0 ({}) at this point in DeleteSession", session_id, session->GetRefCount());
         }
     } else {
         spdlog::warn("Could not delete session {}: not found!", session_id);
@@ -51,9 +58,9 @@ void SessionManager::OnUpgrade(
         return;
     }
 
-    _session_number++;
-    // protect against overflow
-    _session_number = max(_session_number, 1u);
+    struct timeval tv;
+    gettimeofday(&tv, nullptr);
+    _session_number = (((uint32_t)tv.tv_sec) << 16) + ((uint32_t)tv.tv_usec);
 
     http_response->template upgrade<PerSocketData>({_session_number, address}, //
         http_request->getHeader("sec-websocket-key"),                          //
@@ -97,6 +104,7 @@ void SessionManager::OnDisconnect(WSType* ws, int code, std::string_view message
     uint32_t session_id = static_cast<PerSocketData*>(ws->getUserData())->session_id;
 
     // Delete the Session
+    _sessions[session_id]->DecreaseRefCount();
     DeleteSession(session_id);
 
     // Close the websockets
@@ -134,6 +142,7 @@ void SessionManager::OnMessage(WSType* ws, std::string_view sv_message, uWS::OpC
             logger::LogReceivedEventType(event_type);
 
             auto event_type_name = CARTA::EventType_Name(CARTA::EventType(event_type));
+
             bool message_parsed(false);
             OnMessageTask* tsk = nullptr;
 
@@ -148,6 +157,7 @@ void SessionManager::OnMessage(WSType* ws, std::string_view sv_message, uWS::OpC
                 }
                 case CARTA::EventType::RESUME_SESSION: {
                     CARTA::ResumeSession message;
+                    spdlog::debug("({})({}) resuming session", fmt::ptr(session), session->GetId());
                     if (message.ParseFromArray(event_buf, event_length)) {
                         session->OnResumeSession(message, head.request_id);
                         message_parsed = true;

src/Session/SessionManager.h

@@ -20,7 +20,7 @@ class SessionManager {
 public:
     using WSType = uWS::WebSocket<false, true, PerSocketData>;
     SessionManager(ProgramSettings& settings, std::string auth_token, std::shared_ptr<FileListHandler>);
-    void DeleteSession(int session_id);
+    void DeleteSession(uint32_t session_id);
     void OnUpgrade(uWS::HttpResponse<false>* http_response, uWS::HttpRequest* http_request, struct us_socket_context_t* context);
     // Called on connection. Creates session objects and assigns UUID to it
     void OnConnect(WSType* ws);

src/ThreadingManager/ThreadingManager.cc

@@ -45,9 +45,10 @@ void ThreadManager::StartEventHandlingThreads(int num_threads) {
         do {
             std::unique_lock<std::mutex> lock(_task_queue_mtx);
 
-            if (_task_queue.empty() || !(tsk = _task_queue.front())) {
+            if (_task_queue.empty()) {
                 _task_queue_cv.wait(lock);
             } else {
+                tsk = _task_queue.front();
                 _task_queue.pop_front();
                 lock.unlock();
                 tsk->execute();