BastliBridge

manager.d (6689B)

---

 module bastlibridge.manager;
 import bastlibridge.util;
 import bastlibridge.base;
 import std.format;
 import core.thread;
 import core.sync.rwmutex;
 import std.array;
 import std.exception;
 import std.range;
 import std.stdio;
 import std.concurrency;
 import std.algorithm;
 import std.experimental.logger;
 
 
 
 class Manager{
 	
 	enum Control{
 		SHUTDOWN
 	}
 	
 	struct Process{
 		Endpoint endpoint;
 		Thread thread;
 		this(Endpoint ep){
 			endpoint=ep;
 			thread=new Thread(&ep.thread);
 		}
 		void stop(){
 			endpoint.stop();
 		}
 		void start(){
 			tryNTimes!((){endpoint.open();},(Exception e){warning(e.toString); Thread.sleep(1.dur!"seconds");})(5);
 			thread.start();
 		}
 	}
 	
 	private Process[string] endpoints;
 	
 	private LookupTable LUT;
 	
 	private ReadWriteMutex LUT_lock;
 	private ReadWriteMutex endpoints_lock;
 	public string savefile;
 	
 	package __gshared Tid TID;
 	
 	this(){
 		LUT_lock=new ReadWriteMutex();
 		endpoints_lock=new ReadWriteMutex();
 		TID=thisTid();
 	}
 	
 	/**
 	 * Syntax: name=basetype:argstring
 	 */
 	Process addEndpoint(string cmdline){
 		auto ep=createEndpoint(cmdline);
 		auto proc=Process(ep);
 		addProcess(proc);
 		return proc;
 	}
 	
 	void addProcess(Process p){
 		synchronized(endpoints_lock.writer){
 			endpoints[p.endpoint.name]=p;
 		}
 	}
 	
 	Endpoint createEndpoint(string cmdline){
 		string cmdlinetmp=cmdline;
 		string name,basetype,args;
 		cmdline.formattedRead!"%s=%s:%s"(name,basetype,args);
 		if(name in endpoints){
 			throw new Exception(format("Endpoint %s already defined", name));
 		}
 		auto ep=endpointTypes.get(this, basetype, args);
 		ep.name=name;
 		ep.cmdline=cmdlinetmp;
 		return ep;
 	}
 	
 	void dumpConfig(scope void delegate(in char[] s) writer){
 		synchronized(endpoints_lock.reader){
 			foreach(ep; endpoints.byValue){
 				writer(ep.endpoint.cmdline);
 				writer("\n");
 			}
 		}
 		writer("\n");
 		LUT.dumpConfig(writer);
 	}
 	
 	void distribute(Port src, Message msg){
 		synchronized(LUT_lock.reader){
 			LUT[src]
 				.each!(a=>a.send(msg));
 		}
 	}
 	
 	auto getEndpoint(in char[] name){
 		synchronized(endpoints_lock.reader){
 			return name in endpoints;
 		}
 	}
 	
 	void run(){
 		trace("Firing up the threads");
 		endpoints.byValue.each!(a=>a.start());
 	}
 	
 	auto soft_kill(){
 		teardown();
 		Thread.sleep(1000.dur!"msecs");
 	}
 	
 	void handleExit(string name){
 		auto p=getEndpoint(name);
 		p.thread.join();
 		if(p.endpoint.restart){
 			auto process=addEndpoint(p.endpoint.cmdline);
 			replaceEndpoint(name, process);
 			process.start();
 		}
 		else{
 			removeEndpoint(name);
 		}
 	}
 	
 	void serve(){
 		trace("waiting for messages");
 		bool finished=false;
 		try{
 			while(!finished){
 				receive(
 					(string name){
 						trace("Manager is terminating thread ", name);
 						handleExit(name);
 						trace("Current endpoint count ", endpoints.length);
 					},
 					(Manager.Control mc){
 						if(mc==Control.SHUTDOWN){
 							info("Shutting down due to illegal endpoint exit");
 							soft_kill();
 							finished=true;
 						}
 					}
 				);
 				if(endpoints.length==0){
 					finished=true;
 				}
 			}
 		}
 		catch(Exception e){
 			error("Main thread dun goofed ", e.toString);
 			teardown();
 			foreach(p; endpoints){
 				p.thread.join();
 			}
 			
 			return;
 		}
 		info("All processes stopped. We will now exit");
 	}
 	
 	void terminate(in char[] name){
 		auto val=getEndpoint(name);
 		if(val){
 			val.stop();
 		}
 	}
 	
 	void disconnectEndpoint(in char[] name){
 		synchronized(endpoints_lock.reader){
 			synchronized(LUT_lock.writer){
 				foreach(port; endpoints[name].endpoint.ports){
 					LUT.disconnectAll(port);
 				}
 			}
 		}
 	}
 	
 	void reconnectEndpoint(in char[] name, Endpoint newep){
 		synchronized(endpoints_lock.reader){
 			synchronized(LUT_lock.writer){
 				foreach(port; endpoints[name].endpoint.ports){
 					LUT.reconnectAll(port, Port(newep, port.chan));
 				}
 			}
 		}
 	}
 	
 	void removeEndpoint(in char[] name){
 		disconnectEndpoint(name);
 		trace("Removing endpoint ", name);
 		synchronized(endpoints_lock.writer){
 			endpoints.remove(cast(string)name);
 		}
 	}
 	
 	void replaceEndpoint(string name, Process newproc){
 		reconnectEndpoint(name,newproc.endpoint);
 		trace("Reconnecting endpoint ", name);
 		synchronized(endpoints_lock.writer){
 			endpoints[name]=newproc;
 		}
 	}
 	
 	void teardown(){
 		synchronized(endpoints_lock.reader){
 			endpoints.byValue.each!(a=>a.endpoint.stop());
 		}
 	}
 	
 	Port getPort(bool create=false)(in char[] a)
 	out(p){
 		static if(create){
 			assert(p.valid());
 		}
 	}
 	do{
 		auto s=a.findSplit(":");
 		if(s[2].length==0){
 			throw new Exception(format(`Endpoint specifier "%s" malformed`, a));
 		}
 		
 		Process* ep=enforce(getEndpoint(s[0]), format("Endpoint %s not found", s[0]));
 		static if(create){
 			return ep.endpoint.getNewPort(s[2]);
 		}
 		else{
 			return ep.endpoint.getPort(s[2]);
 		}
 	}
 	
 	
 	void save(string file){
 		info("Saving configuration into ", file);
 		File f=File(file, "w+");
 		scope(exit)f.close();
 		dumpConfig((in char[] t)=>f.write(t));
 	}
 	
 	static string LUTMap(string func1, string func2, bool create=false)(){
 		return format!(`
 			void %s(TA, TB)(TA a, TB b){
 					Port gp(T)(T t){
 						static if(is(T == Port)){
 							return t;
 						}
 						else{
 							return getPort!(%s)(t);
 						}
 					}
 					Port pa=gp(a),pb=gp(b);
 					assert(pa.valid(), "pa");
 					assert(pb.valid(), "pb");
 					trace("Doing a %s on ", a, " and ", b);
 				synchronized(LUT_lock.writer){
 					LUT.%s(pa,pb);
 				}
 				if(savefile)
 					save(savefile);
 			}`)(func1, create ? "true" : "false", func2, func2);
 	}
 	mixin(LUTMap!("link", "connect", true)());
 	mixin(LUTMap!("linkDirected", "connectDirected", true)());
 	mixin(LUTMap!("unlink", "disconnect", false)());
 	mixin(LUTMap!("unlinkDirected", "disconnectDirected", false)());
 }
 
 struct LookupTable{
 	
 	private Port[][Port] lut;
 
 	void connect(Port a, Port b){
 		connectDirected(a,b);
 		connectDirected(b,a);
 	}
 	
 	void connectDirected(Port a, Port b){
 		lut[a]~=b;
 	}
 	
 	void disconnectAll(Port a){
 		lut.remove(a);
 		foreach(ref ports; lut.byValue){
 			ports=ports.remove!(p=>p==a)();
 		}
 	}
 	
 	void reconnectAll(Port a, Port b){
 		lut[b]=lut[a];
 		foreach(ref ports; lut.byValue){
 			if(ports.canFind(a)){
 				ports~=b;
 			}
 		}
 		disconnectAll(a);
 	}
 	
 	void disconnect(Port a, Port b){
 		disconnectDirected(a,b);
 		disconnectDirected(b,a);
 	}
 	
 	void disconnectDirected(Port a, Port b){
 		auto i=a in lut;
 		if(!i){
 			return;
 		}
 		*i=remove!(a=>a==b)(*i);
 	}
 	
 	void dumpConfig(scope void delegate(in char[] s) writer){
 		foreach(port; lut.byPair){
 			auto from=port[0];
 			foreach(to; port[1]){
 				formattedWrite(writer,"linkDirected %s %s\n", from.toString, to.toString);
 			}
 		}
 	}
 	
 	auto opIndex(Port a){
 		return lut[a];
 	}
 }