Merge pull request #25 from sunsingerus/master

pool chain

Author: sunsingerus

src/cliopts.py

@@ -58,6 +58,12 @@ def config():
             default=60,
             help='max seconds num between flushes'
         )
+        argparser.add_argument(
+            '--csvpool-file-path-prefix',
+            type=str,
+            default=None,
+            help='file path prefix to CSV pool files'
+        )
 
         argparser.add_argument(
             '--src-server-id',
@@ -173,6 +179,7 @@ def config():
                 'mempool': args.mempool,
                 'mempool-max-events-num': args.mempool_max_events_num,
                 'mempool-max-flush-interval': args.mempool_max_flush_interval,
+                'csvpool_file_path_prefix': args.csvpool_file_path_prefix,
             },
 
             'reader-config': {

src/pool.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import time
+
+from .pool import Pool
+
+
+# Buckets Belts' Index Builder
+class BBIndexBuilder(object):
+
+    def build(self, item):
+        # build key of the belt on which to place item
+        return str(item.schema) + '.' + str(item.table)
+
+
+# Buckets Belts Pool
+class BBPool(Pool):
+
+    # buckets on the belts
+    belts = {
+#                  pour data into 0-index bucket
+#       'key.1': [[item,], [item, item, item,], [item, item, item,]]
+#       'key.2': [[item,], [item, item, item,], [item, item, item,]]
+    }
+
+    belts_rotated_at = {
+#        'key.1': UNIX TIMESTAMP
+#        'key.2': UNIX TIMESTAMP
+    }
+
+    def __init__(
+            self,
+            writer_class=None,
+            writer_params={},
+            key_builder_class=None,
+            max_bucket_size=10000,
+            max_belt_size=1,
+            max_interval_between_rotations=60,
+    ):
+        super().__init__(
+            writer_class=writer_class,
+            writer_params=writer_params,
+            key_builder_class=BBIndexBuilder,
+            max_bucket_size=max_bucket_size,
+            max_belt_size=max_belt_size,
+            max_interval_between_rotations=max_interval_between_rotations,
+        )
+
+    def create_belt(self, belt_index):
+        # create belt with one empty bucket
+        self.belts[belt_index] = [[]]
+        self.belts_rotated_at[belt_index] = int(time.time())
+
+    def insert(self, item):
+        # which belt we'll insert item?
+        belt_index = self.key_builder.build(item)
+
+        # register belt if not yet
+        if belt_index not in self.belts:
+            self.create_belt(belt_index)
+
+        # append item to the 0-indexed bucket of the specified belt
+        self.belts[belt_index][0].append(item)
+
+        # may be bucket is already full
+        if len(self.belts[belt_index][0]) >= self.max_bucket_size:
+            # bucket full, rotate the belt
+            self.rotate_belt(belt_index)
+
+    def flush(self, key=None):
+        belt_index = key
+        empty_belts_indexes = []
+
+        if belt_index is None:
+            for b_index in self.belts:
+                if self.rotate_belt(b_index, flush=True):
+                    empty_belts_indexes.append(b_index)
+        else:
+            if self.rotate_belt(belt_index, flush=True):
+                empty_belts_indexes.append(belt_index)
+
+        # delete belt
+        for b_index in empty_belts_indexes:
+            self.belts.pop(b_index)
+            self.belts_rotated_at.pop(b_index)
+
+    def rotate_belt(self, belt_index, flush=False):
+        now = int(time.time())
+        need_rotation = True if flush else False
+        rotate_by = "FLUSH"
+
+        if len(self.belts[belt_index][0]) >= self.max_bucket_size:
+            # 0-index bucket is full
+            need_rotation = True
+            rotate_by = "SIZE"
+
+        elif now >= self.belts_rotated_at[belt_index] + self.max_interval_between_rotations:
+            # time interval reached
+            need_rotation = True
+            rotate_by = "TIME"
+
+        if not need_rotation:
+            # belt not rotated
+            return False
+
+        # belts needs rotation
+
+        # insert empty bucket into the beginning of the belt
+        self.belts[belt_index].insert(0, [])
+        self.belts_rotated_at[belt_index] = now
+
+        # in case we flush belt we'll keep one just inserted empty bucket
+        buckets_num_left_on_belt = 1 if flush else self.max_belt_size
+
+        while len(self.belts[belt_index]) > buckets_num_left_on_belt:
+            # too many buckets on the belt
+
+            buckets_num = len(self.belts[belt_index])
+            last_bucket_size = len(self.belts[belt_index][buckets_num-1])
+            print(now, 'rotating belt', belt_index, 'rotate by', rotate_by, 'buckets_num', buckets_num, 'last bucket size', last_bucket_size, 'belts:', len(self.belts))
+
+            # time to flush data for specified key
+            writer = self.writer_class(**self.writer_params)
+            writer.insert(self.belts[belt_index].pop())
+            del writer
+
+        # belt rotated
+        return True

src/pool/bbpool.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+
+class Pool(object):
+
+    writer_class = None
+    writer_params = None
+
+    key_builder_class = None
+    key_builder = None
+
+    max_bucket_size = None
+    max_belt_size = None
+    max_interval_between_rotations = None
+
+    def __init__(
+            self,
+            writer_class=None,
+            writer_params={},
+            key_builder_class=None,
+            max_bucket_size=10000,
+            max_belt_size=1,
+            max_interval_between_rotations=60,
+    ):
+        self.writer_class = writer_class
+        self.writer_params = writer_params
+
+        self.key_builder_class = key_builder_class
+        self.key_builder = self.key_builder_class()
+
+        self.max_bucket_size = max_bucket_size
+        self.max_belt_size = max_belt_size
+        self.max_interval_between_rotations = max_interval_between_rotations
+
+    def insert(self, item):
+        pass
+
+    def flush(self, key=None):
+        pass

src/pool/pool.py

@@ -27,7 +27,7 @@ def write_rows_event(self, event=None):
         pass
 
     def write_rows_event_each_row(self, event=None):
-        self.writer.insert(event=event)
+        self.writer.insert(event)
 
     def reader_idle_event(self):
         self.writer.flush()

src/pumper.py

@@ -74,6 +74,9 @@ def read(self):
                         event.table = mysql_event.table
                         self.fire('WriteRowsEvent', event=event)
                         for row in mysql_event.rows:
+                            event = Event()
+                            event.schema = mysql_event.schema
+                            event.table = mysql_event.table
                             event.row = row['values']
                             self.fire('WriteRowsEvent.EachRow', event=event)
                     else:
@@ -85,11 +88,15 @@ def read(self):
 
                 # blocking
                 self.fire('ReaderIdleEvent')
+                time.sleep(1)
 
         except KeyboardInterrupt:
             pass
 
-        self.binlog_stream.close()
+        try:
+            self.binlog_stream.close()
+        except:
+            pass
         end_timestamp = int(time.time())
 
         print('start', start_timestamp)