PNG  IHDR pHYs   OiCCPPhotoshop ICC profilexڝSgTS=BKKoR RB&*! J!QEEȠQ, !{kּ> H3Q5 B.@ $pd!s#~<<+"x M0B\t8K@zB@F&S`cbP-`'{[! eDh;VEX0fK9-0IWfH  0Q){`##xFW<+*x<$9E[-qWW.(I+6aa@.y24x6_-"bbϫp@t~,/;m%h^ uf@Wp~<5j>{-]cK'Xto(hw?G%fIq^D$.Tʳ?D*A, `6B$BB dr`)B(Ͱ*`/@4Qhp.U=pa( Aa!ڈbX#!H$ ɈQ"K5H1RT UH=r9\F;2G1Q= C7F dt1r=6Ыhڏ>C03l0.B8, c˱" VcϱwE 6wB aAHXLXNH $4 7 Q'"K&b21XH,#/{C7$C2'ITFnR#,4H#dk9, +ȅ3![ b@qS(RjJ4e2AURݨT5ZBRQ4u9̓IKhhitݕNWGw Ljg(gwLӋT071oUX**| J&*/Tު UUT^S}FU3S ԖUPSSg;goT?~YYLOCQ_ cx,!k u5&|v*=9C3J3WRf?qtN (~))4L1e\kXHQG6EYAJ'\'GgSSݧ M=:.kDwn^Loy}/TmG X $ <5qo</QC]@Caaᄑ.ȽJtq]zۯ6iܟ4)Y3sCQ? 0k߬~OCOg#/c/Wװwa>>r><72Y_7ȷOo_C#dz%gA[z|!?:eAAA!h쐭!ΑiP~aa~ 'W?pX15wCsDDDޛg1O9-J5*>.j<74?.fYXXIlK9.*6nl {/]py.,:@LN8A*%w% yg"/6шC\*NH*Mz쑼5y$3,幄'L Lݛ:v m2=:1qB!Mggfvˬen/kY- BTZ(*geWf͉9+̳ې7ᒶKW-X潬j9(xoʿܔĹdff-[n ڴ VE/(ۻCɾUUMfeI?m]Nmq#׹=TR+Gw- 6 U#pDy  :v{vg/jBFS[b[O>zG499?rCd&ˮ/~јѡ򗓿m|x31^VwwO| (hSЧc3- cHRMz%u0`:o_F@8N ' p @8N@8}' p '#@8N@8N pQ9p!i~}|6-ӪG` VP.@*j>[ K^<֐Z]@8N'KQ<Q(`s" 'hgpKB`R@Dqj '  'P$a ( `D$Na L?u80e J,K˷NI'0eݷ(NI'؀ 2ipIIKp`:O'`ʤxB8Ѥx Ѥx $ $P6 :vRNb 'p,>NB 'P]-->P T+*^h& p '‰a ‰ (ĵt#u33;Nt̵'ޯ; [3W ~]0KH1q@8]O2]3*̧7# *p>us p _6]/}-4|t'|Smx= DoʾM×M_8!)6lq':l7!|4} '\ne t!=hnLn (~Dn\+‰_4k)0e@OhZ`F `.m1} 'vp{F`ON7Srx 'D˸nV`><;yMx!IS钦OM)Ե٥x 'DSD6bS8!" ODz#R >S8!7ّxEh0m$MIPHi$IvS8IN$I p$O8I,sk&I)$IN$Hi$I^Ah.p$MIN$IR8I·N "IF9Ah0m$MIN$IR8IN$I 3jIU;kO$ɳN$+ q.x* tEXtComment

Viewing File: /opt/cloudlinux/venv/lib/python3.11/site-packages/smmap/test/test_buf.py

from .lib import TestBase, FileCreator

from smmap.mman import (
    SlidingWindowMapManager,
    StaticWindowMapManager
)
from smmap.buf import SlidingWindowMapBuffer

from random import randint
from time import time
import sys
import os


man_optimal = SlidingWindowMapManager()
man_worst_case = SlidingWindowMapManager(
    window_size=TestBase.k_window_test_size // 100,
    max_memory_size=TestBase.k_window_test_size // 3,
    max_open_handles=15)
static_man = StaticWindowMapManager()


class TestBuf(TestBase):

    def test_basics(self):
        with FileCreator(self.k_window_test_size, "buffer_test") as fc:

            # invalid paths fail upon construction
            c = man_optimal.make_cursor(fc.path)
            self.assertRaises(ValueError, SlidingWindowMapBuffer, type(c)())            # invalid cursor
            self.assertRaises(ValueError, SlidingWindowMapBuffer, c, fc.size)       # offset too large

            buf = SlidingWindowMapBuffer()                                              # can create uninitailized buffers
            assert buf.cursor() is None

            # can call end access any time
            buf.end_access()
            buf.end_access()
            assert len(buf) == 0

            # begin access can revive it, if the offset is suitable
            offset = 100
            assert buf.begin_access(c, fc.size) == False
            assert buf.begin_access(c, offset) == True
            assert len(buf) == fc.size - offset
            assert buf.cursor().is_valid()

            # empty begin access keeps it valid on the same path, but alters the offset
            assert buf.begin_access() == True
            assert len(buf) == fc.size
            assert buf.cursor().is_valid()

            # simple access
            with open(fc.path, 'rb') as fp:
                data = fp.read()
            assert data[offset] == buf[0]
            assert data[offset:offset * 2] == buf[0:offset]

            # negative indices, partial slices
            assert buf[-1] == buf[len(buf) - 1]
            assert buf[-10:] == buf[len(buf) - 10:len(buf)]

            # end access makes its cursor invalid
            buf.end_access()
            assert not buf.cursor().is_valid()
            assert buf.cursor().is_associated()         # but it remains associated

            # an empty begin access fixes it up again
            assert buf.begin_access() == True and buf.cursor().is_valid()
            del(buf)        # ends access automatically
            del(c)

            assert man_optimal.num_file_handles() == 1

            # PERFORMANCE
            # blast away with random access and a full mapping - we don't want to
            # exaggerate the manager's overhead, but measure the buffer overhead
            # We do it once with an optimal setting, and with a worse manager which
            # will produce small mappings only !
            max_num_accesses = 100
            fd = os.open(fc.path, os.O_RDONLY)
            for item in (fc.path, fd):
                for manager, man_id in ((man_optimal, 'optimal'),
                                        (man_worst_case, 'worst case'),
                                        (static_man, 'static optimal')):
                    buf = SlidingWindowMapBuffer(manager.make_cursor(item))
                    assert manager.num_file_handles() == 1
                    for access_mode in range(2):    # single, multi
                        num_accesses_left = max_num_accesses
                        num_bytes = 0
                        fsize = fc.size

                        st = time()
                        buf.begin_access()
                        while num_accesses_left:
                            num_accesses_left -= 1
                            if access_mode:  # multi
                                ofs_start = randint(0, fsize)
                                ofs_end = randint(ofs_start, fsize)
                                d = buf[ofs_start:ofs_end]
                                assert len(d) == ofs_end - ofs_start
                                assert d == data[ofs_start:ofs_end]
                                num_bytes += len(d)
                                del d
                            else:
                                pos = randint(0, fsize)
                                assert buf[pos] == data[pos]
                                num_bytes += 1
                            # END handle mode
                        # END handle num accesses

                        buf.end_access()
                        assert manager.num_file_handles()
                        assert manager.collect()
                        assert manager.num_file_handles() == 0
                        elapsed = max(time() - st, 0.001)  # prevent zero division errors on windows
                        mb = float(1000 * 1000)
                        mode_str = (access_mode and "slice") or "single byte"
                        print("%s: Made %i random %s accesses to buffer created from %s reading a total of %f mb in %f s (%f mb/s)"
                              % (man_id, max_num_accesses, mode_str, type(item), num_bytes / mb, elapsed, (num_bytes / mb) / elapsed),
                              file=sys.stderr)
                    # END handle access mode
                    del buf
                # END for each manager
            # END for each input
            os.close(fd)
Back to Directory=ceiIENDB`