<commit_before>'''
Tools for persisting msgpack compatible objects.
'''
import time
import struct
import msgpack
import logging
import threading
import collections

import synapse.compat as s_compat
import synapse.eventbus as s_eventbus
import synapse.telepath as s_telepath

import synapse.lib.queue as s_queue
import synapse.lib.urlhelp as s_urlhelp

from synapse.common import *

logger = logging.getLogger(__name__)

def opendir(*paths, **opts):
    '''
    Open a persistance directory by path name with options.
    '''
    path = gendir(*paths)
    return Dir(path,**opts)

megabyte = 1024000
gigabyte = 1024000000

blocksize = megabyte * 10

class Offset(s_eventbus.EventBus):
    '''
    A file backed persistant offset calculator.

    Example:

        poff = Offset(dirpath,'foo.off'):

        for off,item in pers.items():
            dostuff(item)
            poff.set(off)

    '''
    def __init__(self, *paths):
        s_eventbus.EventBus.__init__(self)

        self.fd = genfile(*paths)
        self.valu = 0

        byts = self.fd.read(8)
        if len(byts):
            self.valu, = struct.unpack('<Q', byts)

        self.onfini( self.fd.close )

    def set(self, valu):
        '''
        Set (and save) the current offset.
        '''
        self.valu = valu

        self.fd.seek(0)
        self.fd.write( struct.pack('<Q',valu) )

        self.fire('pers:off', valu=valu)

    def get(self):
        '''
        Return the current offset.
        '''
        return self.valu

class Dir(s_eventbus.EventBus):
    '''
    A persistance dir may be used similar to a Write-Ahead-Log to sync
    objects based on events ( and allow desync-catch-up )
    '''

    def __init__(self, path, **opts):
        s_eventbus.EventBus.__init__(self)

        self.size = 0
        self.opts = opts
        self.last = None
        self.path = gendir(path)
        self.lock = threading.Lock()


        self.window = collections.deque()

        self.opts.setdefault('filemax',gigabyte) 

        self.baseoff = opts.get('base',0)  # base address of this file
        self.nextoff = self.baseoff + self.size

        self.files = []
        self.pumps = {}
        self.queues = []

        self._initPersFiles()

        if self.files:
            self.last = self.files[-1]

        if self.last == None:
            self.last = self._addPersFile(0)

        self.onfini( self._onDirFini )

    def pump(self, iden, func):
        '''
        Fire a new pump thread to call the given function.

        Example:

            pdir.firePumpThread( iden, core.sync )

        '''
        if self.isfini:
            raise IsFini()

        with self.lock:
            if self.pumps.get(iden):
                raise Exception('Duplicate Pump Iden: %s' % (iden,))

        #self.pumps.append( self._runPumpThread(iden,func) )
        self._runPumpThread(iden,func)

    def getPumpOffs(self):
        '''
        Return a dict of { iden:offset } info for the running pumps.

        Example:

            for iden,noff in pdir.getPumpOffs():
                dostuff()

        '''
        return [ (iden,poff.get()) for (iden,poff) in self.pumps ]

    @firethread
    def _runPumpThread(self, iden, func):
        '''
        Fire a mirror thread to push persist events to a function.
        '''
        poff = Offset(self.path, '%s.off' % iden)

        while not self.isfini:
            noff = poff.get()
            self.pumps[iden] = noff
            try:

                for noff,item in self.items(noff):

                    func(item)
                    poff.set(noff)
                    self.pumps[iden] = noff

                    if self.isfini:
                        return

            except Exception as e:
                if not self.isfini:
                    logger.warning('_runPumpThread (%s): %e' % (iden,e))
                    time.sleep(1)

    def _onDirFini(self):
        [ q.fini() for q in self.queues ]
        [ f.fini() for f in self.files ]

    def _initPersFiles(self):
        # initialize the individual persist files we already have...
        names = [ n for n in os.listdir(self.path) if n.endswith('.cyto') ]
        for name in sorted(names):
            off = int(name.split('.',1)[0],16)
            pers = self._addPersFile(off)
<commit_msg>added getIdenOffset for persist dir to track offsets if desireable<commit_after>'''
Tools for persisting msgpack compatible objects.
'''
import time
import struct
import msgpack
import logging
import threading
import collections

import synapse.compat as s_compat
import synapse.eventbus as s_eventbus
import synapse.telepath as s_telepath

import synapse.lib.queue as s_queue
import synapse.lib.urlhelp as s_urlhelp

from synapse.common import *

logger = logging.getLogger(__name__)

def opendir(*paths, **opts):
    '''
    Open a persistance directory by path name with options.
    '''
    path = gendir(*paths)
    return Dir(path,**opts)

megabyte = 1024000
gigabyte = 1024000000

blocksize = megabyte * 10

class Offset(s_eventbus.EventBus):
    '''
    A file backed persistant offset calculator.

    Example:

        poff = Offset(dirpath,'foo.off'):

        for off,item in pers.items():
            dostuff(item)
            poff.set(off)

    '''
    def __init__(self, *paths):
        s_eventbus.EventBus.__init__(self)

        self.fd = genfile(*paths)
        self.valu = 0

        byts = self.fd.read(8)
        if len(byts):
            self.valu, = struct.unpack('<Q', byts)

        self.onfini( self.fd.close )

    def set(self, valu):
        '''
        Set (and save) the current offset.
        '''
        self.valu = valu

        self.fd.seek(0)
        self.fd.write( struct.pack('<Q',valu) )

        self.fire('pers:off', valu=valu)

    def get(self):
        '''
        Return the current offset.
        '''
        return self.valu

class Dir(s_eventbus.EventBus):
    '''
    A persistance dir may be used similar to a Write-Ahead-Log to sync
    objects based on events ( and allow desync-catch-up )
    '''

    def __init__(self, path, **opts):
        s_eventbus.EventBus.__init__(self)

        self.size = 0
        self.opts = opts
        self.last = None
        self.path = gendir(path)
        self.lock = threading.Lock()


        self.window = collections.deque()

        self.opts.setdefault('filemax',gigabyte) 

        self.baseoff = opts.get('base',0)  # base address of this file
        self.nextoff = self.baseoff + self.size

        self.files = []
        self.pumps = {}
        self.queues = []

        self._initPersFiles()

        if self.files:
            self.last = self.files[-1]

        if self.last == None:
            self.last = self._addPersFile(0)

        self.onfini( self._onDirFini )

    def pump(self, iden, func):
        '''
        Fire a new pump thread to call the given function.

        Example:

            pdir.firePumpThread( iden, core.sync )

        '''
        if self.isfini:
            raise IsFini()

        with self.lock:
            if self.pumps.get(iden):
                raise Exception('Duplicate Pump Iden: %s' % (iden,))

        self._runPumpThread(iden,func)

    def getPumpOffs(self):
        '''
        Return a dict of { iden:offset } info for the running pumps.

        Example:

            for iden,noff in pdir.getPumpOffs():
                dostuff()

        '''
        return [ (iden,poff.get()) for (iden,poff) in self.pumps ]

    def getIdenOffset(self, iden):
        return Offset(self.path, '%s.off' % iden)

    @firethread
    def _runPumpThread(self, iden, func):
        '''
        Fire a mirror thread to push persist events to a function.
        '''
        poff = self.getIdenOffset(iden)

        while not self.isfini:
            noff = poff.get()
            self.pumps[iden] = noff
            try:

                for noff,item in self.items(noff):

                    func(item)
                    poff.set(noff)
                    self.pumps[iden] = noff

                    if self.isfini:
                        return

            except Exception as e:
                if not self.isfini:
                    logger.warning('_runPumpThread (%s): %e' % (iden,e))
                    time.sleep(1)

    def _onDirFini(self):
        [ q.fini() for q in self.queues ]
        [ f.fini() for f in self.files ]

    def _initPersFiles(self):
        # initialize the individual persist files we already have...
        names = [ n for n in os.listdir(self.path) if n.endswith('.cyto') ]
        for name in sorted(names):
            off = int(name.split('.',1)[0],16)
            pers = self._addPersFile(off)

{'fraction_non_alphanumeric': 0.1106764841233318, 'fraction_numerical': 0.007132995858260469, 'mean_word_length': 1.8251543711407214, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3063463', 'n_tokens_mistral': 2819, 'n_tokens_neox': 2642, 'n_words': 686}