Commit 8457a6a9 authored by Kirill Smelkov's avatar Kirill Smelkov

.

parent c00d94c7
...@@ -172,7 +172,7 @@ class DFile: ...@@ -172,7 +172,7 @@ class DFile:
# tDB provides database/wcfs testing environment. # tDB provides database/wcfs testing environment.
# #
# BigFiles opened under tDB are represented as tFile - see .open for details. # BigFiles opened under tDB are represented as tFile - see .open for details.
# Watches opened under tDB are represented as tWatch - see .openwatch for details. # Watches opened under tDB are represented as tWatchLink - see .openwatch for details.
# #
# XXX .open -> .topen # XXX .open -> .topen
# XXX .openwatch -> .topenwatch ? # XXX .openwatch -> .topenwatch ?
...@@ -213,9 +213,9 @@ class tDB: ...@@ -213,9 +213,9 @@ class tDB:
return tFile(t, zf, at=at) return tFile(t, zf, at=at)
# openwatch opens /head/watch on wcfs. # openwatch opens /head/watch on wcfs.
# see returned tWatch for details. # see returned tWatchLink for details.
def openwatch(t): # -> tWatch def openwatch(t): # -> tWatchLink
return tWatch(t) return tWatchLink(t)
# change schedules zf to be changed according changeDelta at commit. # change schedules zf to be changed according changeDelta at commit.
# #
...@@ -338,7 +338,7 @@ class tDB: ...@@ -338,7 +338,7 @@ class tDB:
# iter_revv iterates through all possible at_i -> at_j -> at_k ... sequences. # iter_revv iterates through all possible at_i -> at_j -> at_k ... sequences.
# at_i < at_j # at_i < at_j
# XXX all sequences go till head. # NOTE all sequences go till head.
def iter_revv(t, start=z64, level=0): def iter_revv(t, start=z64, level=0):
dFtail = [_ for _ in t.dFtail if _.rev > start] dFtail = [_ for _ in t.dFtail if _.rev > start]
#print(' '*level, 'iter_revv', t.hat(start), [t.hat(_.rev) for _ in dFtail]) #print(' '*level, 'iter_revv', t.hat(start), [t.hat(_.rev) for _ in dFtail])
...@@ -455,13 +455,11 @@ class tFile: ...@@ -455,13 +455,11 @@ class tFile:
t.assertCache([1]*len(datav)) t.assertCache([1]*len(datav))
# tWatch provides testing environment for /head/watch link opened on wcfs. # tWatchLink provides testing environment for /head/watch link opened on wcfs.
# #
# .sendReq()/.recvReq() provides raw IO in terms of wcfs invalidation protocol messages. # .sendReq()/.recvReq() provides raw IO in terms of wcfs invalidation protocol messages.
# .watch() setups a watch for a file and verifies ... XXX # .watch() setups a watch for a file and verifies ... XXX
# class tWatchLink:
# XXX -> tWatchLink ?
class tWatch:
def __init__(t, tdb): def __init__(t, tdb):
t.tdb = tdb t.tdb = tdb
...@@ -489,9 +487,9 @@ class tWatch: ...@@ -489,9 +487,9 @@ class tWatch:
serveCtx, t._serveCancel = context.with_cancel(context.background()) serveCtx, t._serveCancel = context.with_cancel(context.background())
t._serveWG = sync.WorkGroup(serveCtx) t._serveWG = sync.WorkGroup(serveCtx)
t._serveWG.go(t._serveRecv) t._serveWG.go(t._serveRX)
# this tWatch currently watches files at particular state. # this tWatchLink currently watches files at particular state.
t._watching = {} # {} ZBigFile -> @at t._watching = {} # {} ZBigFile -> @at
tdb._tracked.add(t) tdb._tracked.add(t)
...@@ -501,7 +499,7 @@ class tWatch: ...@@ -501,7 +499,7 @@ class tWatch:
t._serveCancel() t._serveCancel()
# ask wcfs to close its tx & rx sides; close(wcfs.tx) wakes up # ask wcfs to close its tx & rx sides; close(wcfs.tx) wakes up
# _serveRecv on client (= on us). # _serveRX on client (= on us).
t._send(1, b'bye') t._send(1, b'bye')
# XXX we can get stuck here if wcfs does not behave as we want. # XXX we can get stuck here if wcfs does not behave as we want.
# XXX in particular if there is a silly - e.g. syntax or type error in # XXX in particular if there is a silly - e.g. syntax or type error in
...@@ -521,9 +519,9 @@ class tWatch: ...@@ -521,9 +519,9 @@ class tWatch:
# ---- message IO ---- # ---- message IO ----
# _serveRecv receives messages from .w and dispatches them according to streamID. # _serveRX receives messages from .wrx and dispatches them according to streamID.
@func @func
def _serveRecv(t, ctx): def _serveRX(t, ctx):
# when finishing - wakeup everyone waiting for rx # when finishing - wakeup everyone waiting for rx
def _(): def _():
t._acceptq.close() t._acceptq.close()
...@@ -621,16 +619,16 @@ class tWatch: ...@@ -621,16 +619,16 @@ class tWatch:
return tSrvReq(t, stream, msg) return tSrvReq(t, stream, msg)
# tSrvReq represents 1 server-initiated wcfs request received over /head/watch. # tSrvReq represents 1 server-initiated wcfs request received over /head/watch link.
class tSrvReq: class tSrvReq:
def __init__(req, twatch, stream, msg): def __init__(req, twlink, stream, msg):
req.twatch = twatch req.twlink = twlink
req.stream = stream req.stream = stream
req.msg = msg req.msg = msg
def reply(req, answer): def reply(req, answer):
#print('C: reply %s <- %r ...' % (req, answer)) #print('C: reply %s <- %r ...' % (req, answer))
t = req.twatch t = req.twlink
with t._rxmu: with t._rxmu:
assert req.stream in t._accepted assert req.stream in t._accepted
...@@ -648,10 +646,10 @@ class tSrvReq: ...@@ -648,10 +646,10 @@ class tSrvReq:
# _pinAt returns which blocks needs to be pinned for zf@at. # _pinAt returns which blocks needs to be pinned for zf@at.
# #
# it does not take into account whether blocks are in cache or not and computes # it does not take into account whether blocks are in cache or not and computes
# pin from all changes. XXX desired behaviour? # pin from all changes. XXX is it desired behaviour?
@func(tWatch) @func(tWatchLink)
def _pinAt(w, zf, at): # -> pin = {} blk -> rev def _pinAt(twlink, zf, at): # -> pin = {} blk -> rev
t = w.tdb t = twlink.tdb
# all changes to zf # all changes to zf
vdf = [_.byfile[zf] for _ in t.dFtail if zf in _.byfile] vdf = [_.byfile[zf] for _ in t.dFtail if zf in _.byfile]
...@@ -676,10 +674,10 @@ def _pinAt(w, zf, at): # -> pin = {} blk -> rev ...@@ -676,10 +674,10 @@ def _pinAt(w, zf, at): # -> pin = {} blk -> rev
# watch sets up a watch for file@at. # watch sets up a watch for file@at.
# XXX and verifies that wcfs sends correct initial pins? # XXX and verifies that wcfs sends correct initial pins?
# XXX or adjusts # XXX or adjusts
@func(tWatch) @func(tWatchLink)
def watch(w, zf, at): # XXX -> ? def watch(twlink, zf, at): # XXX -> ?
t = w.tdb t = twlink.tdb
at_prev = w._watching.get(zf) # we were previously watching zf @at_prev at_prev = twlink._watching.get(zf) # we were previously watching zf @at_prev
at_from = '' at_from = ''
if at_prev is not None: if at_prev is not None:
at_from = '(%s ->) ' % t.hat(at_prev) at_from = '(%s ->) ' % t.hat(at_prev)
...@@ -700,9 +698,9 @@ def watch(w, zf, at): # XXX -> ? ...@@ -700,9 +698,9 @@ def watch(w, zf, at): # XXX -> ?
pin_prev = {} pin_prev = {}
if at_prev is not None: if at_prev is not None:
assert at_prev <= at, 'TODO %s -> %s' % (t.hat(at_prev), t.hat(at)) assert at_prev <= at, 'TODO %s -> %s' % (t.hat(at_prev), t.hat(at))
pin_prev = w._pinAt(zf, at_prev) pin_prev = twlink._pinAt(zf, at_prev)
pin = w._pinAt(zf, at) pin = twlink._pinAt(zf, at)
if at_prev != at and at_prev is not None: if at_prev != at and at_prev is not None:
print('# pin@old: %s\n# pin@new: %s' % (pinstr(pin_prev), pinstr(pin))) print('# pin@old: %s\n# pin@new: %s' % (pinstr(pin_prev), pinstr(pin)))
...@@ -743,12 +741,12 @@ def watch(w, zf, at): # XXX -> ? ...@@ -743,12 +741,12 @@ def watch(w, zf, at): # XXX -> ?
wg = sync.WorkGroup(ctx) wg = sync.WorkGroup(ctx)
def _(ctx): def _(ctx):
pinv = w._expectPin(ctx, zf, pinok) pinv = twlink._expectPin(ctx, zf, pinok)
for p in pinv: for p in pinv:
p.reply(b"ack") # XXX -> return to caller? p.reply(b"ack") # XXX -> return to caller?
# check that we don't get extra pins before "ok" reply to "watch" # check that we don't get extra pins before "ok" reply to "watch"
try: try:
req = w.recvReq(ctx) req = twlink.recvReq(ctx)
except Exception as e: except Exception as e:
if e is context.canceled: if e is context.canceled:
return # cancel is expected after seeing "ok" return # cancel is expected after seeing "ok"
...@@ -758,21 +756,21 @@ def watch(w, zf, at): # XXX -> ? ...@@ -758,21 +756,21 @@ def watch(w, zf, at): # XXX -> ?
wg.go(_) wg.go(_)
def _(ctx): def _(ctx):
assert w.sendReq(ctx, b"watch %s @%s" % (h(zf._p_oid), h(at))) == "ok" assert twlink.sendReq(ctx, b"watch %s @%s" % (h(zf._p_oid), h(at))) == "ok"
# cancel _expectPin waiting upon receiving "ok" from wcfs # cancel _expectPin waiting upon receiving "ok" from wcfs
# -> error that missed pins were not received. # -> error that missed pins were not received.
cancel() cancel()
wg.go(_) wg.go(_)
wg.wait() wg.wait()
w._watching[zf] = at twlink._watching[zf] = at
# _expectPin asserts that wcfs sends expected pin messages. # _expectPin asserts that wcfs sends expected pin messages.
# #
# expect is {} blk -> at # expect is {} blk -> at
# returns [] of received pin requests. # returns [] of received pin requests.
@func(tWatch) @func(tWatchLink)
def _expectPin(w, ctx, zf, expect): def _expectPin(twlink, ctx, zf, expect):
expected = set() # of expected pin messages expected = set() # of expected pin messages
for blk, at in expect.items(): for blk, at in expect.items():
hat = h(at) if at is not None else 'head' hat = h(at) if at is not None else 'head'
...@@ -783,7 +781,7 @@ def _expectPin(w, ctx, zf, expect): ...@@ -783,7 +781,7 @@ def _expectPin(w, ctx, zf, expect):
reqv = [] # of received requests reqv = [] # of received requests
while len(expected) > 0: while len(expected) > 0:
try: try:
req = w.recvReq(ctx) req = twlink.recvReq(ctx)
except Exception as e: except Exception as e:
raise RuntimeError("%s\nnot all pin missages received - pending:\n%s" % (e, expected)) raise RuntimeError("%s\nnot all pin missages received - pending:\n%s" % (e, expected))
assert req is not None # channel not closed assert req is not None # channel not closed
...@@ -877,26 +875,23 @@ def test_wcfs(): ...@@ -877,26 +875,23 @@ def test_wcfs():
# XXX invalid requests -> wcfs replies error # XXX invalid requests -> wcfs replies error
# XXX -> separate test? # XXX -> separate test?
w = t.openwatch() wl = t.openwatch()
print('\n\nzzzzzzz\n\n') assert wl.sendReq(context.background(), b'bla bla') == ""
assert w.sendReq(context.background(), b'bla bla') == ""
# assert w closed # assert wl closed
print('\n\n0000000\n\n') wl.close()
w.close()
print('\n\n1111111\n\n')
for zf in t.zfiles(): for zf in t.zfiles():
# watch going at_i -> at_j -> ... # watch going at_i -> at_j -> ...
for revv in t.iter_revv(): for revv in t.iter_revv():
print('\n--------') print('\n--------')
#print(' -> '.join([t.hat(_) for _ in revv])) #print(' -> '.join([t.hat(_) for _ in revv]))
w = t.openwatch() wl = t.openwatch()
w.watch(zf, revv[0]) wl.watch(zf, revv[0])
w.watch(zf, revv[0]) # verify at_i -> at_i wl.watch(zf, revv[0]) # verify at_i -> at_i
for at in revv[1:]: for at in revv[1:]:
w.watch(zf, at) wl.watch(zf, at)
w.close() wl.close()
print() print()
...@@ -941,22 +936,6 @@ def test_wcfs(): ...@@ -941,22 +936,6 @@ def test_wcfs():
# setupWatch must send pins. # setupWatch must send pins.
@func
def test_wcfs_invproto():
# XXX temp debug
import sys, traceback
def _():
print('BBB')
1/0
xdefer(_)
#raise RuntimeError('zzz')
print('\nAAA')
assert 1 == 2
# ---- misc --- # ---- misc ---
# readfile reads file @ path. # readfile reads file @ path.
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment