spawn io worker for drive subsystem

.circleci/config.yml

@@ -58,7 +58,7 @@ commands:
           name: run tests
           command: |
             . venv/bin/activate
-            mkdir test-reports
+            mkdir -p test-reports
             circleci tests glob synapse/tests/test_*.py synapse/vendor/**/test_*.py | \
               circleci tests run \
                 --timings-type=name \
@@ -458,7 +458,8 @@ jobs:
           CODECOV_FLAG: linux
           SYN_REGRESSION_REPO: ~/git/synapse-regression
           COVERAGE_FILE: test-reports/<< pipeline.git.revision >>/.coverage
-          COVERAGE_ARGS: --cov synapse --cov-append
+          COVERAGE_PROCESS_START: .coveragerc
+          COVERAGE_ARGS: --cov synapse --cov-config=.coveragerc.main --cov-append
 
     working_directory: ~/repo
 
@@ -474,7 +475,8 @@ jobs:
           RUN_SYNTAX: 1
           CODECOV_FLAG: linux_replay
           SYN_REGRESSION_REPO: ~/git/synapse-regression
-          COVERAGE_ARGS: --cov synapse
+          COVERAGE_PROCESS_START: .coveragerc
+          COVERAGE_ARGS: --cov synapse --cov-config=.coveragerc.main
           SYNDEV_NEXUS_REPLAY: 1
 
     working_directory: ~/repo

.coveragerc

@@ -1,12 +1,17 @@
 [report]
 omit =
     */synapse/tests/test_*
+    /tmp/*
+
+[run]
+concurrency = multiprocessing
+parallel = True
+sigterm = True
 
 ; Uncomment this section to enable code coverage of storm files in the
 ; storm_dirs directory listed below. This is disabled by default right now
 ; because it's pretty intensive and imposes a large perf hit on the already slow
 ; tests.
-;[run]
 ;plugins = synapse.utils.stormcov
 
 [synapse.utils.stormcov]

.coveragerc.main

@@ -0,0 +1,6 @@
+[report]
+omit =
+    */synapse/tests/test_*
+
+[run]
+sigterm = True

changes/c56856d0929e4a71fde0d98fc77ddeb5.yaml

@@ -0,0 +1,6 @@
+---
+desc: Migrated cell drive data into a dedicated slab.
+desc:literal: false
+prs: []
+type: migration
+...

changes/fb1de9c8091f353422a50736baf4d803.yaml

@@ -0,0 +1,7 @@
+---
+desc: Added a dedicated IO worker for the drive subsystem to offload operations into
+  a separate process.
+desc:literal: false
+prs: []
+type: feat
+...

synapse/lib/cell.py

@@ -2,6 +2,7 @@
 import os
 import ssl
 import copy
+import stat
 import time
 import fcntl
 import shutil
@@ -1181,6 +1182,8 @@ async def __anit__(self, dirn, conf=None, readonly=False, parent=None):
         s_telepath.Aware.__init__(self)
 
         self.dirn = s_common.gendir(dirn)
+        self.sockdirn = s_common.gendir(dirn, 'sockets')
+
         self.runid = s_common.guid()
 
         self.auth = None
@@ -1521,14 +1524,34 @@ async def _storCellAuthMigration(self):
         logger.warning(f'...Cell ({self.getCellType()}) auth migration complete!')
 
     async def _drivePermMigration(self):
-        for lkey, lval in self.slab.scanByPref(s_drive.LKEY_INFO, db=self.drive.dbname):
-            info = s_msgpack.un(lval)
-            perm = info.pop('perm', None)
-            if perm is not None:
-                perm.setdefault('users', {})
-                perm.setdefault('roles', {})
-                info['permissions'] = perm
-                self.slab.put(lkey, s_msgpack.en(info), db=self.drive.dbname)
+        async with await s_drive.Drive.anit(self.slab, 'celldrive') as olddrive:
+            for lkey, lval in self.slab.scanByPref(s_drive.LKEY_INFO, db=olddrive.dbname):
+                info = s_msgpack.un(lval)
+                perm = info.pop('perm', None)
+                if perm is not None:
+                    perm.setdefault('users', {})
+                    perm.setdefault('roles', {})
+                    info['permissions'] = perm
+                    self.slab.put(lkey, s_msgpack.en(info), db=olddrive.dbname)
+
+    async def _driveCellMigration(self):
+        logger.warning('Migrating Drive Slabs')
+
+        self.olddrive = await s_drive.Drive.anit(self.slab, 'celldrive')
+
+        dbname = self.olddrive.dbname
+        newpath = s_common.gendir(self.dirn, 'slabs', 'drive.lmdb')
+
+        async with await s_lmdbslab.Slab.anit(newpath) as newslab:
+            rows = await self.slab.copydb(dbname, newslab, dbname)
+            logger.warning(f"Migrated {rows} rows")
+            newslab.forcecommit()
+
+        await self.olddrive.fini()
+        self.slab.dropdb(dbname)
+        self.olddrive = None
+
+        logger.warning('...Drive migration complete!')
 
     def getPermDef(self, perm):
         perm = tuple(perm)
@@ -1672,22 +1695,34 @@ def _delTmpFiles(self):
         tdir = s_common.gendir(self.dirn, 'tmp')
 
         names = os.listdir(tdir)
-        if not names:
-            return
+        if names:
+            logger.warning(f'Removing {len(names)} temporary files/folders in: {tdir}')
 
-        logger.warning(f'Removing {len(names)} temporary files/folders in: {tdir}')
+            for name in names:
 
-        for name in names:
+                path = os.path.join(tdir, name)
 
-            path = os.path.join(tdir, name)
+                if os.path.isfile(path):
+                    os.unlink(path)
+                    continue
 
-            if os.path.isfile(path):
-                os.unlink(path)
-                continue
+                if os.path.isdir(path):
+                    shutil.rmtree(path, ignore_errors=True)
+                    continue
 
-            if os.path.isdir(path):
-                shutil.rmtree(path, ignore_errors=True)
-                continue
+        names = os.listdir(self.sockdirn)
+        if names:
+            logger.info(f'Removing {len(names)} old sockets in: {self.sockdirn}')
+            for name in names:
+                path = os.path.join(self.sockdirn, name)
+                try:
+                    if stat.S_ISSOCK(os.stat(path).st_mode):
+                        os.unlink(path)
+                except OSError: # pragma: no cover
+                    pass
+
+        shutil.rmtree(self.sockdirn, ignore_errors=True)
+        self.sockdirn = s_common.gendir(self.dirn, 'sockets')
 
     async def _execCellUpdates(self):
         # implement to apply updates to a fully initialized active cell
@@ -1899,11 +1934,20 @@ async def initServiceEarly(self):
         pass
 
     async def initCellStorage(self):
-        self.drive = await s_drive.Drive.anit(self.slab, 'celldrive')
         await self._bumpCellVers('drive:storage', (
             (1, self._drivePermMigration),
+            (2, self._driveCellMigration),
         ), nexs=False)
 
+        path = s_common.gendir(self.dirn, 'slabs', 'drive.lmdb')
+        sockpath = s_common.genpath(self.sockdirn, 'drive')
+
+        if len(sockpath) > s_const.UNIX_PATH_MAX:
+            sockpath = None
+
+        spawner = s_drive.FileDrive.spawner(base=self, sockpath=sockpath)
+        self.drive = await spawner(path)
+
         self.onfini(self.drive.fini)
 
     async def addDriveItem(self, info, path=None, reldir=s_drive.rootdir):
@@ -1922,7 +1966,7 @@ async def _addDriveItem(self, info, path=None, reldir=s_drive.rootdir):
 
         # replay safety...
         iden = info.get('iden')
-        if self.drive.hasItemInfo(iden): # pragma: no cover
+        if await self.drive.hasItemInfo(iden): # pragma: no cover
             return await self.drive.getItemPath(iden)
 
         # TODO: Remove this in synapse-3xx
@@ -1935,10 +1979,10 @@ async def _addDriveItem(self, info, path=None, reldir=s_drive.rootdir):
         return await self.drive.addItemInfo(info, path=path, reldir=reldir)
 
     async def getDriveInfo(self, iden, typename=None):
-        return self.drive.getItemInfo(iden, typename=typename)
+        return await self.drive.getItemInfo(iden, typename=typename)
 
-    def reqDriveInfo(self, iden, typename=None):
-        return self.drive.reqItemInfo(iden, typename=typename)
+    async def reqDriveInfo(self, iden, typename=None):
+        return await self.drive.reqItemInfo(iden, typename=typename)
 
     async def getDrivePath(self, path, reldir=s_drive.rootdir):
         '''
@@ -1961,15 +2005,14 @@ async def addDrivePath(self, path, perm=None, reldir=s_drive.rootdir):
         '''
         tick = s_common.now()
         user = self.auth.rootuser.iden
-        path = self.drive.getPathNorm(path)
+        path = await self.drive.getPathNorm(path)
 
         if perm is None:
             perm = {'users': {}, 'roles': {}}
 
         for name in path:
 
-            info = self.drive.getStepInfo(reldir, name)
-            await asyncio.sleep(0)
+            info = await self.drive.getStepInfo(reldir, name)
 
             if info is not None:
                 reldir = info.get('iden')
@@ -1992,20 +2035,20 @@ async def getDriveData(self, iden, vers=None):
         Return the data associated with the drive item by iden.
         If vers is specified, return that specific version.
         '''
-        return self.drive.getItemData(iden, vers=vers)
+        return await self.drive.getItemData(iden, vers=vers)
 
     async def getDriveDataVersions(self, iden):
         async for item in self.drive.getItemDataVersions(iden):
             yield item
 
     @s_nexus.Pusher.onPushAuto('drive:del')
     async def delDriveInfo(self, iden):
-        if self.drive.getItemInfo(iden) is not None:
+        if await self.drive.getItemInfo(iden) is not None:
             await self.drive.delItemInfo(iden)
 
     @s_nexus.Pusher.onPushAuto('drive:set:perm')
     async def setDriveInfoPerm(self, iden, perm):
-        return self.drive.setItemPerm(iden, perm)
+        return await self.drive.setItemPerm(iden, perm)
 
     @s_nexus.Pusher.onPushAuto('drive:data:path:set')
     async def setDriveItemProp(self, iden, vers, path, valu):
@@ -2054,9 +2097,9 @@ async def delDriveItemProp(self, iden, vers, path):
     @s_nexus.Pusher.onPushAuto('drive:set:path')
     async def setDriveInfoPath(self, iden, path):
 
-        path = self.drive.getPathNorm(path)
+        path = await self.drive.getPathNorm(path)
         pathinfo = await self.drive.getItemPath(iden)
-        if path == [p.get('name') for p in pathinfo]:
+        if [str(p) for p in path] == [p.get('name', '') for p in pathinfo]:
             return pathinfo
 
         return await self.drive.setItemPath(iden, path)
@@ -2067,13 +2110,13 @@ async def setDriveData(self, iden, versinfo, data):
 
     async def delDriveData(self, iden, vers=None):
         if vers is None:
-            info = self.drive.reqItemInfo(iden)
+            info = await self.drive.reqItemInfo(iden)
             vers = info.get('version')
         return await self._push('drive:data:del', iden, vers)
 
     @s_nexus.Pusher.onPush('drive:data:del')
     async def _delDriveData(self, iden, vers):
-        return self.drive.delItemData(iden, vers)
+        return await self.drive.delItemData(iden, vers)
 
     async def getDriveKids(self, iden):
         async for info in self.drive.getItemKids(iden):

synapse/lib/const.py

@@ -51,3 +51,6 @@
 # HTTP header constants
 MAX_LINE_SIZE = kibibyte * 64
 MAX_FIELD_SIZE = kibibyte * 64
+
+# Socket constants
+UNIX_PATH_MAX = 107