Add logic to support ranged GTID updates from proxysql_mysqlbinlog.

include/proxysql_gtid.h

@@ -2,12 +2,38 @@
 #define PROXYSQL_GTID
 // highly inspired by libslave
 // https://github.com/vozbu/libslave/
-#include <unordered_map>
 #include <list>
+#include <string>
+#include <unordered_map>
 #include <utility>
 
 typedef std::pair<std::string, int64_t> gtid_t;
-typedef std::pair<int64_t, int64_t> gtid_interval_t;
+
+class Gtid_Interval {
+	public:
+		int64_t start;
+		int64_t end;
+
+	private:
+		void init(const int64_t _start, const int64_t _end);
+		void init(const char* s);
+
+	public:
+		explicit Gtid_Interval(const int64_t _start, const int64_t _end);
+		explicit Gtid_Interval(const char* s);
+		explicit Gtid_Interval(const std::string& s);
+
+		const std::string to_string(void);
+		const bool contains(int64_t gtid);
+		const bool merge(const Gtid_Interval& other);
+
+		const int cmp(const Gtid_Interval& other);
+		const bool operator<(const Gtid_Interval& other);
+		const bool operator==(const Gtid_Interval& other);
+};
+typedef Gtid_Interval gtid_interval_t;
+
+// TODO: make me a proper class.
 typedef std::unordered_map<std::string, std::list<gtid_interval_t>> gtid_set_t;
 
 /*
@@ -30,4 +56,4 @@
 };
 */
 
-#endif /* PROXYSQL_GTID */
+#endif /* PROXYSQL_GTID */

lib/GTID_Server_Data.cpp

@@ -58,7 +58,7 @@
 			if (it2 != MyHGM->gtid_map.end()) {
 				//MyHGM->gtid_map.erase(it2);
 				it2->second = NULL;
 				delete sd;
 			}
 			ev_io_stop(MyHGM->gtid_ev_loop, w);
 			free(w);
@@ -93,10 +93,10 @@
 			if (it2 != MyHGM->gtid_map.end()) {
 				//MyHGM->gtid_map.erase(it2);
 				it2->second = NULL;
 				delete sd;
 			}
 			//delete custom_data;
 			free(c);
 		} else {
 			ev_io_stop(MyHGM->gtid_ev_loop, w);
 			int fd=w->fd;
@@ -149,7 +149,7 @@
 		struct ev_io *c = (struct ev_io *)malloc(sizeof(struct ev_io));
 		if (c) {
 			ev_io_init(c, connect_cb, s, EV_WRITE);
 			GTID_Server_Data * custom_data = new GTID_Server_Data(c, address, gtid_port, mysql_port);
 			c->data = (void *)custom_data;
 			return c;
 		}
@@ -220,7 +220,7 @@
 		return false;
 	}
 	for (auto itr = it->second.begin(); itr != it->second.end(); ++itr) {
-		if ((int64_t)gtid_trxid >= itr->first && (int64_t)gtid_trxid <= itr->second) {
+		if (itr->contains((int64_t)gtid_trxid)) {
 //			fprintf(stderr,"YES\n");
 			return true;
 		}
@@ -375,12 +375,7 @@
 		s.insert(23,"-");
 		s = s + ":";
 		for (auto itr = it->second.begin(); itr != it->second.end(); ++itr) {
-			std::string s2 = s;
-			s2 = s2 + std::to_string(itr->first);
-			s2 = s2 + "-";
-			s2 = s2 + std::to_string(itr->second);
-			s2 = s2 + ",";
-			gtid_set = gtid_set + s2;
+			gtid_set += s + itr->to_string() + ",";
 		}
 	}
 	// Extract latest comma only in case 'gtid_executed' isn't empty
@@ -391,54 +386,47 @@
 }
 
 
-
-void addGtid(const gtid_t& gtid, gtid_set_t& gtid_executed) {
-	auto it = gtid_executed.find(gtid.first);
-	if (it == gtid_executed.end())
-	{
-		gtid_executed[gtid.first].emplace_back(gtid.second, gtid.second);
+// Merges a GTID interval into a gitd_executed instance.
+void addGtid(const std::string& uuid, const gtid_interval_t &iv, gtid_set_t& gtid_executed) {
+	auto it = gtid_executed.find(uuid);
+	if (it == gtid_executed.end()) {
+		// new UUID entry
+		gtid_executed[uuid].emplace_back(iv);
 		return;
 	}
 
-	bool flag = true;
-	for (auto itr = it->second.begin(); itr != it->second.end(); ++itr)
-	{
-		if (gtid.second >= itr->first && gtid.second <= itr->second)
-			return;
-		if (gtid.second + 1 == itr->first)
-		{
-			--itr->first;
-			flag = false;
-			break;
-		}
-		else if (gtid.second == itr->second + 1)
-		{
-			++itr->second;
-			flag = false;
+	// insert/merge GTID interval
+	auto pos = it->second.begin();
+	for (; pos != it->second.end(); ++pos) {
+		if (pos->merge(iv))
 			break;
-		}
-		else if (gtid.second < itr->first)
-		{
-			it->second.emplace(itr, gtid.second, gtid.second);
-			return;
-		}
+	}
+	if (pos == it->second.end()) {
+		it->second.emplace_back(iv);
 	}
 
-	if (flag)
-		it->second.emplace_back(gtid.second, gtid.second);
-
-	for (auto itr = it->second.begin(); itr != it->second.end(); ++itr)
-	{
-		auto next_itr = std::next(itr);
-		if (next_itr != it->second.end() && itr->second + 1 == next_itr->first)
-		{
-			itr->second = next_itr->second;
-			it->second.erase(next_itr);
+	// merge overlapping GTID ranges, if any
+	it->second.sort();
+	auto a = it->second.begin();
+	while (a != it->second.end()) {
+		auto b = std::next(a);
+		if (b == it->second.end()) {
 			break;
 		}
+		if (a->merge(*b)) {
+				it->second.erase(b);
+				continue;
+		}
+		a++;
 	}
 }
 
+// Merges a single GTID into a gitd_executed instance.
+inline void addGtid(const gtid_t& gtid, gtid_set_t& gtid_executed) {
+	gtid_interval_t iv = Gtid_Interval(gtid.second, gtid.second);
+	addGtid(gtid.first, iv, gtid_executed);
+}
+
 void * GTID_syncer_run() {
 	//struct ev_loop * gtid_ev_loop;
 	//gtid_ev_loop = NULL;

lib/Makefile

@@ -71,6 +71,7 @@ _OBJ_CXX := ProxySQL_GloVars.oo network.oo debug.oo configfile.oo Query_Cache.oo
 	ProxySQL_Admin_Tests.oo ProxySQL_Admin_Tests2.oo ProxySQL_Admin_Scheduler.oo ProxySQL_Admin_Disk_Upgrade.oo ProxySQL_Admin_Stats.oo \
 	Admin_Handler.oo Admin_FlushVariables.oo Admin_Bootstrap.oo \
 	Base_Session.oo Base_Thread.oo \
+	proxysql_gtid.oo \
 	proxy_protocol_info.oo \
 	proxysql_find_charset.oo ProxySQL_Poll.oo \
 	PgSQL_Protocol.oo PgSQL_Thread.oo PgSQL_Data_Stream.oo PgSQL_Session.oo PgSQL_Variables.oo PgSQL_HostGroups_Manager.oo PgSQL_Connection.oo PgSQL_Backend.oo PgSQL_Logger.oo PgSQL_Authentication.oo PgSQL_Error_Helper.oo \

lib/proxysql_gtid.cpp

@@ -0,0 +1,111 @@
+#include <cstdint>
+#include <cstdio>
+#include <cstdlib>
+#include <string>
+
+#include "proxysql_gtid.h"
+
+
+// Initializes a GTID interval.
+// Implemented via private method as C++03 does not support delegated constructors :(
+void Gtid_Interval::init(const int64_t _start, const int64_t _end) {
+	start = _start;
+	end = _end;
+
+	if (start > end) {
+		std::swap(start, end);
+	}
+}
+
+// Initializes a GTID interval from a string buffer, in [gtid]{-[gtid]} format.
+void Gtid_Interval::init(const char *s) {
+	uint64_t _start, _end;
+
+	if (sscanf(s, "%lu-%lu", &_start, &_end) == 2) {
+		init((int64_t)_start, (int64_t)_end);
+		return;
+	}
+	if (sscanf(s, "%lu", &_start) == 1) {
+		init((int64_t)_start, (int64_t)_start);
+		return;
+	}
+
+	init(0, 0);
+}
+
+
+Gtid_Interval::Gtid_Interval(const int64_t _start, const int64_t _end) {
+	init(_start, _end);
+}
+
+Gtid_Interval::Gtid_Interval(const char *s) {
+	init(s);
+}
+
+Gtid_Interval::Gtid_Interval(const std::string& s) {
+	init(s.c_str());
+}
+
+// Checks if a given GTID is contained in this interval.
+const bool Gtid_Interval::contains(int64_t gtid) {
+	return (gtid >= start && gtid <= end);
+}
+
+// Yields a string representation for a GTID interval.
+const std::string Gtid_Interval::to_string(void) {
+	if (start == end) {
+		return std::to_string(start);
+	}
+	return std::to_string(start) + "-" + std::to_string(end);
+}
+
+// Attempts to merge two GTID intervals. Returns true if the intervals were merged (and potentially modified), false otherwise.
+const bool Gtid_Interval::merge(const Gtid_Interval& other) {
+	if (other.start >= start && other.end <= end) {
+		// other is contained by interval
+		return true;
+	}
+	if (other.start <= start && other.end >= end) {
+		// other contains whole of existing interval
+		start = other.start;
+		end = other.end;
+		return true;
+	}
+	if (other.start <= start && other.end >= (start-1)) {
+		// other overlaps interval at start
+		start = other.start;
+		return true;
+	}
+	if (other.end >= end && other.start <= (end+1)) {
+		// other overlaps interval at end
+		end = other.end;
+		return true;
+	}
+
+	return false;
+}
+
+// Comapres two GTID intervals, by strict weak ordering.
+const int Gtid_Interval::cmp(const Gtid_Interval& other) {
+	if (start < other.start) {
+		return -1;
+	}
+	if (start > other.start) {
+		return 1;
+	}
+	if (end < other.end) {
+		return -1;
+	}
+	if (end > other.end) {
+		return 1;
+	}
+	return 0;
+}
+
+const bool Gtid_Interval::operator<(const Gtid_Interval& other) {
+	return cmp(other) == -1;
+}
+
+const bool Gtid_Interval::operator==(const Gtid_Interval& other) {
+	return cmp(other) == 0;
+}