profile I/O in lock

Signed-off-by: KantaTamura <[email protected]>

Author: KantaTamura

pfio/cache/file_cache.py

@@ -234,12 +234,14 @@ def _get(self, i):
 
         offset = self.buflen * i
         with self.lock.rdlock(), record("pfio.cache.file:get:lock", trace=self.trace):
-            buf = os.pread(self.cachefp.fileno(), self.buflen, offset)
+            with record("pfio.cache.file:get:read_index", trace=self.trace):
+                buf = os.pread(self.cachefp.fileno(), self.buflen, offset)
             (o, l) = unpack('Qq', buf)
             if l < 0 or o < 0:
                 return None
 
-            data = os.pread(self.cachefp.fileno(), l, o)
+            with record("pfio.cache.file:get:read_data", trace=self.trace):
+                data = os.pread(self.cachefp.fileno(), l, o)
             assert len(data) == l
             return data
 
@@ -276,7 +278,8 @@ def _put(self, i, data):
 
         offset = self.buflen * i
         with self.lock.wrlock(), record("pfio.cache.file:put:lock", trace=self.trace):
-            buf = os.pread(self.cachefp.fileno(), self.buflen, offset)
+            with record("pfio.cache.file:put:read_index", trace=self.trace):
+                buf = os.pread(self.cachefp.fileno(), self.buflen, offset)
             (o, l) = unpack('Qq', buf)
             if l >= 0 and o >= 0:
                 # Already data exists
@@ -301,13 +304,15 @@ def _put(self, i, data):
 
             '''
             buf = pack('Qq', pos, len(data))
-            r = os.pwrite(self.cachefp.fileno(), buf, offset)
+            with record("pfio.cache.file:put:write_index", trace=self.trace):
+                r = os.pwrite(self.cachefp.fileno(), buf, offset)
             assert r == self.buflen
 
             current_pos = pos
             while current_pos - pos < len(data):
-                r = os.pwrite(self.cachefp.fileno(),
-                              data[current_pos-pos:], current_pos)
+                with record("pfio.cache.file:put:write_data", trace=self.trace):
+                    r = os.pwrite(self.cachefp.fileno(),
+                                  data[current_pos-pos:], current_pos)
                 assert r > 0
                 current_pos += r
             assert current_pos - pos == len(data)

pfio/cache/multiprocess_file_cache.py

@@ -234,13 +234,14 @@ def _get(self, i):
 
         fcntl.flock(self.cache_fd, fcntl.LOCK_SH)
         with record(f"pfio.cache.multiprocessfile:get:lock-{self.cache_fd}", trace=self.trace):
-            index_entry = os.pread(self.cache_fd, self.buflen, offset)
+            with record("pfio.cache.multiprocessfile:get:read_index", trace=self.trace):
+                index_entry = os.pread(self.cache_fd, self.buflen, offset)
             (o, l) = unpack('Qq', index_entry)
             if l < 0 or o < 0:
                 fcntl.flock(self.cache_fd, fcntl.LOCK_UN)
                 return None
 
-            with record("pfio.cache.multiprocessfile:get:read", trace=self.trace):
+            with record("pfio.cache.multiprocessfile:get:read_data", trace=self.trace):
                 data = os.pread(self.cache_fd, l, o)
             assert len(data) == l
             fcntl.flock(self.cache_fd, fcntl.LOCK_UN)
@@ -279,26 +280,30 @@ def _put(self, i, data):
 
         fcntl.flock(self.cache_fd, fcntl.LOCK_EX)
         with record(f"pfio.cache.multiprocessfile:put:lock-{self.cache_fd}", trace=self.trace):
-            buf = os.pread(self.cache_fd, self.buflen, index_ofst)
+            with record("pfio.cache.multiprocessfile:put:read_index", trace=self.trace):
+                buf = os.pread(self.cache_fd, self.buflen, index_ofst)
             (o, l) = unpack('Qq', buf)
 
             if l >= 0 and o >= 0:
                 # Already data exists
                 fcntl.flock(self.cache_fd, fcntl.LOCK_UN)
                 return False
 
-            data_pos = os.lseek(self.cache_fd, 0, os.SEEK_END)
+            with record("pfio.cache.multiprocessfile:put:seek", trace=self.trace):
+                data_pos = os.lseek(self.cache_fd, 0, os.SEEK_END)
             if self.cache_size_limit:
                 if self.cache_size_limit < (data_pos + len(data)):
                     self._frozen = True
                     fcntl.flock(self.cache_fd, fcntl.LOCK_UN)
                     return False
 
             index_entry = pack('Qq', data_pos, len(data))
-            assert os.pwrite(self.cache_fd, index_entry, index_ofst) == self.buflen
-            with record("pfio.cache.multiprocessfile:put:write", trace=self.trace):
+            with record("pfio.cache.multiprocessfile:put:write_index", trace=self.trace):
+                assert os.pwrite(self.cache_fd, index_entry, index_ofst) == self.buflen
+            with record("pfio.cache.multiprocessfile:put:write_data", trace=self.trace):
                 assert os.pwrite(self.cache_fd, data, data_pos) == len(data)
-            os.fsync(self.cache_fd)
+            with record("pfio.cache.multiprocessfile:put:sync", trace=self.trace):
+                os.fsync(self.cache_fd)
             fcntl.flock(self.cache_fd, fcntl.LOCK_UN)
 
         return True