test (48e591ab) · Commits · tuxist / authdb

src/authdb.cpp

+39 −134

Changes for src/authdb.cpp: 39 added lines, 134 removed lines.
Original line number	Diff line number	Diff line
		@@ -31,9 +31,6 @@
		#include <mutex>
		#include <fstream>
		#include <thread>
		#include <queue>
		#include <condition_variable>
		#include <functional>

		#include <cmdplus/cmdplus.h>
		#include <confplus/conf.h>
		@@ -86,73 +83,7 @@ namespace authdb {
		bool initalized=true;
		bool DEBUG=false;

		// Thread pool for offloading blocking work from event handler threads.
		// RequestEvent dispatches blocking operations (backend I/O, cluster calls)
		// to this pool so event loop threads are never blocked.
		class WorkerPool {
		public:
		explicit WorkerPool(size_t num_threads) : stop_(false) {
		for (size_t i = 0; i < num_threads; i++) {
		workers_.emplace_back([this]() { run(); });
		}
		}

		~WorkerPool() {
		{
		std::lock_guard<std::mutex> lock(mutex_);
		stop_ = true;
		}
		cv_.notify_all();
		for (auto& t : workers_) {
		if (t.joinable()) t.join();
		}
		}

		void submit(std::function<void()> task) {
		{
		std::lock_guard<std::mutex> lock(mutex_);
		tasks_.push(std::move(task));
		}
		cv_.notify_one();
		}

		private:
		void run() {
		while (true) {
		std::function<void()> task;
		{
		std::unique_lock<std::mutex> lock(mutex_);
		cv_.wait(lock, [this]() { return stop_ \|\| !tasks_.empty(); });
		if (stop_ && tasks_.empty()) return;
		task = std::move(tasks_.front());
		tasks_.pop();
		}
		task();
		}
		}

		std::vector<std::thread> workers_;
		std::queue<std::function<void()>> tasks_;
		std::mutex mutex_;
		std::condition_variable cv_;
		bool stop_;
		};

		// Helper: send an error response safely from a worker thread.
		// Acquires event_mutex on the connection and decrements PendingOps.
		static void sendErrorFromWorker(libhttppp::HttpRequest &curreq,
		const char *state,
		const char *msg, size_t msglen) {
		{
		std::lock_guard<std::recursive_mutex> lock(curreq.event_mutex);
		libhttppp::HttpResponse rep;
		rep.setState(state);
		rep.setContentType("text/html");
		rep.send(curreq, msg, msglen);
		curreq.slots[0].ResponsePending.store(true);
		}
		curreq.slots[0].PendingOps.fetch_sub(1);
		}

		class AuthDB : public libhttppp::HttpEvent{
		public:
		@@ -275,11 +206,15 @@ namespace authdb {
		}

		void RequestEvent(libhttppp::HttpRequest &curreq, const int tid, ULONG_PTR args){
		const char *methods[]={"PARSE","GET","POST","DELETE","PUT"};
		int rtype=curreq.getRequestType();
		const char *method=(rtype>=0 && rtype<=4) ? methods[rtype] : "UNKNOWN";

		std::string clientAddr;
		if(!curreq.slots.empty() && curreq.slots[0].csock)
		curreq.slots[0].csock->getAddress(clientAddr);

		// Non-blocking: serve static assets directly from memory
		try {
		if(curreq.getRequestURL().compare(0,10,"/authdb.js")==0){
		libhttppp::HttpResponse rep;
		rep.setState(HTTP200);
		@@ -302,37 +237,10 @@ namespace authdb {
		return;
		}

		// Non-blocking: redirect unknown paths
		if(curreq.getRequestURL().compare(0,9,"/settings",9)!=0 &&
		curreq.getRequestURL().compare(0,4,"/api",4)!=0 &&
		curreq.getRequestURL().compare(0,10,"/getavatar",10)!=0 &&
		_AdminBackend.end()>sizeof(authdb::AuthHeader)){
		libhttppp::HttpResponse rep;
		rep.setState(HTTP307);
		rep.setHeaderData("Location")->push_back("/settings/index");
		rep.setContentType("text/html");
		rep.send(curreq,"",0);
		return;
		}

		// All remaining paths involve blocking I/O (backend locks, cluster
		// calls, file scans). Offload to worker pool to keep event threads
		// free for other connections.
		curreq.slots[0].PendingOps.fetch_add(1);

		_pool.submit([this, &curreq, tid, args, clientAddr]() {
		const char *methods[]={"PARSE","GET","POST","DELETE","PUT"};
		int rtype=curreq.getRequestType();
		const char *method=(rtype>=0 && rtype<=4) ? methods[rtype] : "UNKNOWN";

		try {
		if(_AdminBackend.end()<=sizeof(authdb::AuthHeader)){
		std::lock_guard<std::recursive_mutex> conlock(curreq.event_mutex);
		AuthBackendLock admlock(_AdminBackend);
		try{
		WizzardEvent(curreq,tid,args);
		curreq.slots[0].ResponsePending.store(true);
		curreq.slots[0].PendingOps.fetch_sub(1);
		return;
		}catch(libhttppp::HTTPException &e){
		if(e.getErrorType()!=999){
		@@ -340,15 +248,12 @@ namespace authdb {
		rep.setState(HTTP500);
		rep.setContentType("text/html");
		rep.send(curreq,e.what(),e.what().length());
		curreq.slots[0].ResponsePending.store(true);
		curreq.slots[0].PendingOps.fetch_sub(1);
		return;
		}
		}
		}

		if(curreq.getRequestURL().compare(0,9,"/settings",9)==0){
		std::lock_guard<std::recursive_mutex> conlock(curreq.event_mutex);
		AuthBackendLock admlock(_AdminBackend);
		try{
		_AdminInterface.Request(curreq,tid,args);
		@@ -361,23 +266,26 @@ namespace authdb {
		rep.setContentType("text/html");
		rep.send(curreq,e.what(),strlen(e.what()));
		}
		curreq.slots[0].ResponsePending.store(true);
		}else if(curreq.getRequestURL().compare(0,4,"/api",4)==0){
		DomainBackend db(_AdminBackend);

		AuthBackend *apidb=nullptr;

		char domain[255]={0};

		if(sscanf(curreq.getRequestURL().c_str(),"/api/%254[^/]s",domain)<0){
		sendErrorFromWorker(curreq, HTTP500, "can't parse url", 15);
		libhttppp::HttpResponse rep;
		rep.setState(HTTP500);
		rep.setContentType("text/html");
		rep.send(curreq,"can't parse url",15);
		return;
		}

		::uuid::uuid did;

		if(did.parse(domain)!=0){
		sendErrorFromWorker(curreq, HTTP500, "can't parse domain id", 21);
		libhttppp::HttpResponse rep;
		rep.setState(HTTP500);
		rep.setContentType("text/html");
		rep.send(curreq,"can't parse domain id",21);
		return;
		}

		@@ -388,10 +296,16 @@ namespace authdb {
		try {
		dinfo=db.resolve(did.value);
		}catch(std::exception &e){
		sendErrorFromWorker(curreq, HTTP500, e.what(), strlen(e.what()));
		libhttppp::HttpResponse rep;
		rep.setState(HTTP500);
		rep.setContentType("text/html");
		rep.send(curreq,e.what(),strlen(e.what()));
		return;
		}catch(...){
		sendErrorFromWorker(curreq, HTTP500, "internal error loading domain", 29);
		libhttppp::HttpResponse rep;
		rep.setState(HTTP500);
		rep.setContentType("text/html");
		rep.send(curreq,"internal error loading domain",29);
		return;
		}
		}
		@@ -399,15 +313,20 @@ namespace authdb {
		try {
		apidb=db.create(dinfo);
		}catch(std::exception &e){
		sendErrorFromWorker(curreq, HTTP500, e.what(), strlen(e.what()));
		libhttppp::HttpResponse rep;
		rep.setState(HTTP500);
		rep.setContentType("text/html");
		rep.send(curreq,e.what(),strlen(e.what()));
		return;
		}catch(...){
		sendErrorFromWorker(curreq, HTTP500, "internal error creating domain backend", 38);
		libhttppp::HttpResponse rep;
		rep.setState(HTTP500);
		rep.setContentType("text/html");
		rep.send(curreq,"internal error creating domain backend",38);
		return;
		}

		if(apidb){
		std::lock_guard<std::recursive_mutex> conlock(curreq.event_mutex);
		apidb->lock();
		try{
		ApiController(*apidb,did,curreq,tid,args);
		@@ -415,30 +334,18 @@ namespace authdb {
		}catch(std::exception &e){
		g_Log.error(clientAddr,e.what());
		g_Log.access(clientAddr,method,curreq.getRequestURL(),500);
		apidb->unlock();
		curreq.slots[0].ResponsePending.store(true);
		curreq.slots[0].PendingOps.fetch_sub(1);
		return;
		}catch(...){
		g_Log.access(clientAddr,method,curreq.getRequestURL(),500);
		apidb->unlock();
		curreq.slots[0].ResponsePending.store(true);
		curreq.slots[0].PendingOps.fetch_sub(1);
		return;
		}
		apidb->unlock();
		curreq.slots[0].ResponsePending.store(true);
		}else{
		std::lock_guard<std::recursive_mutex> conlock(curreq.event_mutex);
		g_Log.access(clientAddr,method,curreq.getRequestURL(),404);
		libhttppp::HttpResponse rep;
		rep.setState(HTTP404);
		rep.setContentType("text/html");
		rep.send(curreq,"uuid not found!",12);
		curreq.slots[0].ResponsePending.store(true);
		}
		}else if(curreq.getRequestURL().compare(0,10,"/getavatar",10)==0){
		std::lock_guard<std::recursive_mutex> conlock(curreq.event_mutex);
		AuthBackendLock admlock(_AdminBackend);
		try{
		getAvatar(curreq,tid,args);
		@@ -451,27 +358,26 @@ namespace authdb {
		rep.setContentType("text/html");
		rep.send(curreq,e.what(),strlen(e.what()));
		}
		curreq.slots[0].ResponsePending.store(true);
		}else{
		// Wizard initialized while we were queued — redirect
		std::lock_guard<std::recursive_mutex> conlock(curreq.event_mutex);
		if(_AdminBackend.end()>sizeof(authdb::AuthHeader)){
		libhttppp::HttpResponse rep;
		rep.setState(HTTP307);
		rep.setHeaderData("Location")->push_back("/settings/index");
		rep.setContentType("text/html");
		rep.send(curreq,"",0);
		curreq.slots[0].ResponsePending.store(true);
		}
		}
		} catch (std::exception &e) {
		sendErrorFromWorker(curreq, HTTP500, e.what(), strlen(e.what()));
		return;
		libhttppp::HttpResponse rep;
		rep.setState(HTTP500);
		rep.setContentType("text/html");
		rep.send(curreq,e.what(),strlen(e.what()));
		} catch (...) {
		sendErrorFromWorker(curreq, HTTP500, "internal error", 14);
		return;
		libhttppp::HttpResponse rep;
		rep.setState(HTTP500);
		rep.setContentType("text/html");
		rep.send(curreq,"internal error",14);
		}

		curreq.slots[0].PendingOps.fetch_sub(1);
		});
		}

		void getAvatar(libhttppp::HttpRequest &curreq, const int tid, ULONG_PTR args){
		@@ -537,7 +443,6 @@ namespace authdb {
		AdminInterface _AdminInterface;
		std::vector<char> _AuthJs;
		std::vector<char> _AuthCss;
		WorkerPool _pool{16};
		};

		int searchValue(authdb::AuthBackend &backend,const charfieldname,const char value){