[WIP] Experimental . Making mysql_sessions private

include/Base_Thread.h

@@ -1,6 +1,6 @@
 #ifndef CLASS_BASE_THREAD_H
 #define CLASS_BASE_THREAD_H
-
+#include <type_traits>
 #include "proxysql.h"
 
 typedef struct _thr_id_username_t {
@@ -36,47 +36,75 @@ class Session_Regex {
 class MySQL_Thread;
 class PgSQL_Thread;
 
+template<typename T>
 class Base_Thread {
+	static_assert(std::is_same_v<T,MySQL_Thread> || std::is_same_v<T,PgSQL_Thread>,
+		"Invalid Thread type");
 	private:
+	using TypeSession = typename std::conditional<std::is_same_v<T,MySQL_Thread>,MySQL_Session,PgSQL_Session>::type;
+	using TypeDataStream = typename std::conditional<std::is_same_v<T,MySQL_Thread>,MySQL_Data_Stream,PgSQL_Data_Stream>::type;
 	bool maintenance_loop;
+	protected:
+	std::vector<TypeSession *> mysql_sessions;
 	public:
+	std::mutex mysql_sessions_mutex;  // Protect access to mysql_sessions , if needed
 	unsigned long long curtime;
 	unsigned long long last_move_to_idle_thread_time;
 	bool epoll_thread;
 	int shutdown;
-	PtrArray *mysql_sessions;
 	Session_Regex **match_regexes;
 	Base_Thread();
 	~Base_Thread();
-	template<typename T, typename S>
-	S create_new_session_and_client_data_stream(int _fd);
-	template<typename T, typename S>
-	void register_session(T, S, bool up_start = true);
-	template<typename T>
+	TypeSession * create_new_session_and_client_data_stream(int _fd);
+	void register_session(TypeSession *, bool up_start = true);
+	/**
+	 * @brief Unregisters a session from the thread's session array.
+	 *
+	 * @param idx The index of the session to unregister.
+	 * @param lock Whatever the lock should be taken or not
+	 *
+	 * @details This function removes a session from the `mysql_sessions` array at the specified index.
+	 * It does not delete the session object itself; it is assumed that the caller will handle
+	 * the deletion.
+	 *
+	 * @note This function is called by various parts of the code when a session is no longer
+	 * active and needs to be removed from the thread's session list.
+	 *
+	 */
+	void unregister_session(int, bool);
+
+	/**
+	 * @brief Unregisters a session from the thread's session array.
+	 *
+	 * @param sess The address of the Session
+	 * @param lock Whatever the lock should be taken or not
+	 *
+	 * @details This function removes a session from the `mysql_sessions` array.
+	 * It does not delete the session object itself; it is assumed that the caller will handle
+	 * the deletion.
+	 *
+	 * @note This function is called by various parts of the code when a session is no longer
+	 * active and needs to be removed from the thread's session list.
+	 *
+	 */
+	void unregister_session(TypeSession *, bool);
+
 	void check_timing_out_session(unsigned int n);
-	template<typename T>
 	void check_for_invalid_fd(unsigned int n);
-	template<typename S>
 	void ProcessAllSessions_SortingSessions();
-	template<typename T>
 	void ProcessAllMyDS_AfterPoll();
-	template<typename T>
 	void read_one_byte_from_pipe(unsigned int n);
-	template<typename T, typename DS>
-	void tune_timeout_for_myds_needs_pause(DS * myds);
-	template<typename T, typename DS>
-	void tune_timeout_for_session_needs_pause(DS * myds);
-	template<typename T, typename DS>
-	void configure_pollout(DS * myds, unsigned int n);
-	template<typename T, typename DS>
-	bool set_backend_to_be_skipped_if_frontend_is_slow(DS * myds, unsigned int n);
+	void tune_timeout_for_myds_needs_pause(TypeDataStream * myds);
+	void tune_timeout_for_session_needs_pause(TypeDataStream * myds);
+	void configure_pollout(TypeDataStream * myds, unsigned int n);
+	bool set_backend_to_be_skipped_if_frontend_is_slow(TypeDataStream * myds, unsigned int n);
 #ifdef IDLE_THREADS
-	template<typename T, typename DS> bool move_session_to_idle_mysql_sessions(DS * myds, unsigned int n);
+	bool move_session_to_idle_mysql_sessions(TypeDataStream * myds, unsigned int n);
 #endif // IDLE_THREADS
-	template<typename T, typename S> unsigned int find_session_idx_in_mysql_sessions(S * sess);
-	template<typename T> void ProcessAllMyDS_BeforePoll();
-	template<typename T, typename S> void run_SetAllSession_ToProcess0();
-
+	unsigned int find_session_idx_in_mysql_sessions(TypeSession * sess);
+	void ProcessAllMyDS_BeforePoll();
+	void run_SetAllSession_ToProcess0();
+	const std::vector<TypeSession *>& get_mysql_sessions_ref() const { return mysql_sessions; };
 
 #if ENABLE_TIMER
 	// for now this is not accessible via Admin/Prometheus , thus useful only with gdb

include/MySQL_Thread.h

@@ -89,7 +89,7 @@ enum MySQL_Thread_status_variable {
 	MY_st_var_END
 };
 
-class __attribute__((aligned(64))) MySQL_Thread : public Base_Thread
+class __attribute__((aligned(64))) MySQL_Thread : public Base_Thread<MySQL_Thread>
 {
 	friend class PgSQL_Thread;
 	private:
@@ -179,6 +179,7 @@ class __attribute__((aligned(64))) MySQL_Thread : public Base_Thread
 	struct {
 		unsigned long long stvar[MY_st_var_END];
 		unsigned int active_transactions;
+		std::atomic<unsigned int> non_idle_client_connections;
 	} status_variables;
 
 	struct {
@@ -206,7 +207,7 @@ class __attribute__((aligned(64))) MySQL_Thread : public Base_Thread
   void poll_listener_add(int sock);
   void poll_listener_del(int sock);
   //void register_session(MySQL_Session *, bool up_start=true);
-  void unregister_session(int);
+  //void unregister_session(int);
   struct pollfd * get_pollfd(unsigned int i);
   bool process_data_on_data_stream(MySQL_Data_Stream *myds, unsigned int n);
 	//void ProcessAllSessions_SortingSessions();
@@ -216,13 +217,15 @@ class __attribute__((aligned(64))) MySQL_Thread : public Base_Thread
 	void process_all_sessions();
   void refresh_variables();
   void register_session_connection_handler(MySQL_Session *_sess, bool _new=false);
-  void unregister_session_connection_handler(int idx, bool _new=false);
+  //void unregister_session_connection_handler(int idx, bool _new=false);
   void listener_handle_new_connection(MySQL_Data_Stream *myds, unsigned int n);
 	void Get_Memory_Stats();
 	MySQL_Connection * get_MyConn_local(unsigned int, MySQL_Session *sess, char *gtid_uuid, uint64_t gtid_trxid, int max_lag_ms);
 	void push_MyConn_local(MySQL_Connection *);
 	void return_local_connections();
 	void Scan_Sessions_to_Kill(PtrArray *mysess);
+	void Scan_Sessions_to_Kill(const std::vector<MySQL_Session *>& sessions);
+	void Scan_Sessions_to_Kill_innerLoop(MySQL_Session *_sess);
 	void Scan_Sessions_to_Kill_All();
 };
 

include/PgSQL_Thread.h

@@ -137,7 +137,7 @@ struct CopyCmdMatcher {
 	}
 };
 
-class __attribute__((aligned(64))) PgSQL_Thread : public Base_Thread
+class __attribute__((aligned(64))) PgSQL_Thread : public Base_Thread<PgSQL_Thread>
 {
 private:
 	unsigned int servers_table_version_previous;
@@ -231,6 +231,7 @@ class __attribute__((aligned(64))) PgSQL_Thread : public Base_Thread
 	struct {
 		unsigned long long stvar[PG_st_var_END];
 		unsigned int active_transactions;
+		std::atomic<unsigned int> non_idle_client_connections;
 	} status_variables;
 
 	struct {
@@ -392,7 +393,7 @@ class __attribute__((aligned(64))) PgSQL_Thread : public Base_Thread
 	 * active and needs to be removed from the thread's session list.
 	 *
 	 */
-	void unregister_session(int);
+	//void unregister_session(int);
 
 	/**
 	 * @brief Returns a pointer to the `pollfd` structure for a specific data stream.
@@ -535,7 +536,7 @@ class __attribute__((aligned(64))) PgSQL_Thread : public Base_Thread
 	 * removed from the connection handler list.
 	 *
 	 */
-	void unregister_session_connection_handler(int idx, bool _new = false);
+	//void unregister_session_connection_handler(int idx, bool _new = false);
 
 	/**
 	 * @brief Handles a new connection accepted by a listener.
@@ -649,6 +650,10 @@ class __attribute__((aligned(64))) PgSQL_Thread : public Base_Thread
 	 */
 	void Scan_Sessions_to_Kill(PtrArray * mysess);
 
+
+	void Scan_Sessions_to_Kill(const std::vector<PgSQL_Session *>& sessions);
+	void Scan_Sessions_to_Kill_innerLoop(PgSQL_Session *_sess);
+
 	/**
 	 * @brief  Scans all session arrays across all threads to identify and kill sessions.
 	 *

include/gen_utils.h

@@ -271,6 +271,33 @@ class FixedSizeQueue : public std::queue<buffer_t> {
 
 #endif /* __CLASS_PTR_ARRAY_H */
 
+#ifndef CLASS_ConditionalLock_H
+#define CLASS_ConditionalLock_H
+#include <mutex>
+
+class ConditionalLock {
+private:
+	std::mutex& mutex;
+	bool locked;
+
+public:
+	ConditionalLock(std::mutex& m, bool should_lock) : mutex(m), locked(should_lock) {
+		if (locked) {
+			mutex.lock();
+		}
+	}
+
+	~ConditionalLock() {
+		if (locked) {
+			mutex.unlock();
+		}
+	}
+
+	// Prevent copying and assignment
+	ConditionalLock(const ConditionalLock&) = delete;
+	ConditionalLock& operator=(const ConditionalLock&) = delete;
+};
+#endif // CLASS_ConditionalLock_H
 
 
 #ifndef __GEN_FUNCTIONS