wip

mt.py

@@ -0,0 +1,155 @@
+from threading import Thread, Lock
+import time
+
+
+class mt():
+    def __init__(self, threads, data, verbose=False):
+        self.__running = False
+        self.__closed = False
+        self.__data = data
+        self.__verbose = verbose
+        # dummy
+        self.__final = None
+        self.__comment = None
+        self.__starttime = None
+        self.__endtime = None
+        self.__type = None
+        # thread things
+        self.__threadcount = threads
+        self.__threads = []
+        self.__lock = Lock()
+        self.__results = []
+        for i in range(self.__threadcount):
+            self.__results.append([])
+            self.__threads.append(None)
+
+    def filter(self, cond, comment=None):
+        if self.__closed:
+            raise RuntimeError("Already closed")
+        if self.__running:
+            self.join()
+            self.__data = self.getresults()
+        self.__running = True
+        self.__final = self.__getresultsmapfilter
+        self.__type = "filter"
+        self.__comment = comment if comment is not None else ""
+        if comment is not None:
+            print(self.__comment + ": #" + str(len(self.__data)) + " ...", end='\n' if self.__verbose else '', flush=True)
+        self.__starttime = self.__cms()
+        self.__endtime = None
+        for i in range(self.__threadcount):
+            part = self.__data[i * len(self.__data) // self.__threadcount:(i + 1) * len(self.__data) // self.__threadcount]
+            self.__threads[i] = Thread(target=self.__dofilter, args=(i, part, cond))
+            self.__threads[i].start()
+        return self
+
+    def __dofilter(self, i, list, cond):
+        now = self.__cms()
+        results = [l for l in list if cond(l)]
+        with self.__lock:
+            self.__results[i] = results
+        dur = self.__cms() - now
+        if self.__verbose:
+            print(self.__comment + ": Thread " + str(i) + ": filter took " + str(dur) + "ms")
+
+    def map(self, func, comment=None):
+        if self.__closed:
+            raise RuntimeError("Already closed")
+        if self.__running:
+            self.join()
+            self.__data = self.getresults()
+        self.__running = True
+        self.__final = self.__getresultsmapfilter
+        self.__type = "map"
+        self.__comment = comment if comment is not None else ""
+        if comment is not None:
+            print(self.__comment + ": #" + str(len(self.__data)) + " ...", end='\n' if self.__verbose else '', flush=True)
+        self.__starttime = self.__cms()
+        self.__endtime = None
+        for i in range(self.__threadcount):
+            part = self.__data[i * len(self.__data) // self.__threadcount:(i + 1) * len(self.__data) // self.__threadcount]
+            self.__threads[i] = Thread(target=self.__domap, args=(i, part, func))
+            self.__threads[i].start()
+        return self
+
+    def __domap(self, i, list, func):
+        now = self.__cms()
+        results = [func(l) for l in list]
+        with self.__lock:
+            self.__results[i] = results
+        dur = self.__cms() - now
+        if self.__verbose:
+            print(self.__comment + ": Thread " + str(i) + ": map took " + str(dur) + "ms")
+
+    def reduce(self, reducer, aggregator, initval, comment=None):
+        if self.__closed:
+            raise RuntimeError("Already closed")
+        if self.__running:
+            self.join()
+            self.__data = self.getresults()
+        self.__running = True
+        self.__final = lambda: self.__getresultsreduce(aggregator, initval)
+        self.__type = "reduce"
+        self.__comment = comment if comment is not None else ""
+        if comment is not None:
+            print(self.__comment + ": #" + str(len(self.__data)) + " ...", end='\n' if self.__verbose else '', flush=True)
+        self.__starttime = self.__cms()
+        self.__endtime = None
+        for i in range(self.__threadcount):
+            part = self.__data[i * len(self.__data) // self.__threadcount:(i + 1) * len(self.__data) // self.__threadcount]
+            self.__threads[i] = Thread(target=self.__doreduce, args=(i, part, reducer, initval))
+            self.__threads[i].start()
+        return self
+
+    def __doreduce(self, i, list, reducer, initval):
+        now = self.__cms()
+        val = initval()
+        for j in range(len(list)):
+            val = reducer(val, list[j])
+        with self.__lock:
+            self.__results[i] = val
+        dur = self.__cms() - now
+        if self.__verbose:
+            print(self.__comment + ": Thread " + str(i) + ": reduce took " + str(dur) + "ms")
+
+    def getresults(self):
+        self.join()
+        return self.__final()
+
+    def __getresultsmapfilter(self):
+        res = []
+        for i in range(self.__threadcount):
+            res += self.__results[i]
+        return res
+
+    def __getresultsreduce(self, aggregator, initval):
+        val = initval()
+        for j in range(self.__threadcount):
+            val = aggregator(val, self.__results[j])
+        return val
+
+    def join(self):
+        if self.__closed:
+            raise RuntimeError("Already closed")
+        for i in range(self.__threadcount):
+            if self.__threads[i] is not None:
+                self.__threads[i].join()
+                self.__threads[i] = None
+        if self.__endtime is None:
+            self.__endtime = self.__cms();
+            if self.__comment is not None:
+                dur = self.__endtime - self.__starttime
+                if self.__verbose:
+                    print(self.__comment + ": #" + str(len(self.__data)) + (" -> #" + str(sum([len(l) for l in self.__results])) if self.__type == "filter" else "") + " ... took " + str(dur) + "ms")
+                else:
+                    print("\r" + self.__comment + ": #" + str(len(self.__data)) + (" -> #" + str(sum([len(l) for l in self.__results])) if self.__type == "filter" else "") + " ... took " + str(dur) + "ms")
+        return self
+
+    def close(self):
+        if self.__closed:
+            raise RuntimeError("Already closed")
+        self.join()
+        self.__closed = True
+
+    def __cms(self):
+        return int(round(time.time() * 1000))