def receive_objects_v2(conn, junk):
global suspended_w
git.check_repo_or_die()
- suggested = {}
+ suggested = set()
if suspended_w:
w = suspended_w
suspended_w = None
n -= 20 + 4
buf = conn.read(n) # object sizes in bup are reasonably small
#debug2('read %d bytes\n' % n)
- if len(buf) < n:
- w.abort()
- raise Exception('object read: expected %d bytes, got %d\n'
- % (n, len(buf)))
+ _check(w, n, len(buf), 'object read: expected %d bytes, got %d\n')
if not dumb_server_mode:
oldpack = w.exists(shar)
- # FIXME: we only suggest a single index per cycle, because the client
- # is currently too dumb to download more than one per cycle anyway.
- # Actually we should fix the client, but this is a minor optimization
- # on the server side.
- if not suggested and \
- oldpack and (oldpack == True or oldpack.endswith('.midx')):
- # FIXME: we shouldn't really have to know about midx files
- # at this layer. But exists() on a midx doesn't return the
- # packname (since it doesn't know)... probably we should just
- # fix that deficiency of midx files eventually, although it'll
- # make the files bigger. This method is certainly not very
- # efficient.
- oldpack = w.objcache.packname_containing(shar)
- debug2('new suggestion: %r\n' % oldpack)
- assert(oldpack)
- assert(oldpack != True)
- assert(not oldpack.endswith('.midx'))
- w.objcache.refresh()
- if not suggested and oldpack:
+ if oldpack:
+ if oldpack == True or oldpack.endswith('.midx'):
+ # FIXME: we shouldn't really have to know about midx files
+ # at this layer. But exists() on a midx doesn't return the
+ # packname (since it doesn't know)... probably we should
+ # just fix that deficiency of midx files eventually,
+ # although it'll make the files bigger. This method is
+ # certainly not very efficient.
+ oldpack = w.objcache.packname_containing(shar)
+ debug2('new suggestion: %r\n' % oldpack)
+ w.objcache.refresh()
assert(oldpack.endswith('.idx'))
(dir,name) = os.path.split(oldpack)
if not (name in suggested):
debug1("bup server: suggesting index %s\n" % name)
conn.write('index %s\n' % name)
- suggested[name] = 1
+ suggested.add(name)
continue
- nw, crc = w._raw_write([buf], sha=shar)
+ nw, crc = w._raw_write((buf,), sha=shar)
_check(w, crcr, crc, 'object read: expected crc %d, got %d\n')
- _check(w, n, nw, 'object read: expected %d bytes, got %d\n')
# NOTREACHED
def _make_objcache(self):
return git.PackIdxList(self.cachedir)
- def _suggest_pack(self, indexname):
- debug1('client: received index suggestion: %s\n' % indexname)
+ def _suggest_packs(self):
ob = self._busy
if ob:
assert(ob == 'receive-objects-v2')
- self.conn.write('\xff\xff\xff\xff') # suspend receive-objects
+ self.conn.write('\xff\xff\xff\xff') # suspend receive-objects-v2
+ suggested = []
+ for line in linereader(self.conn):
+ if not line:
+ break
+ debug2('%s\n' % line)
+ if line.startswith('index '):
+ idx = line[6:]
+ debug1('client: received index suggestion: %s\n' % idx)
+ suggested.append(idx)
+ else:
+ assert(line.endswith('.idx'))
+ debug1('client: completed writing pack, idx: %s\n' % line)
+ suggested.append(line)
+ self.check_ok()
+ if ob:
self._busy = None
- self.conn.drain_and_check_ok()
- self.sync_index(indexname)
+ for idx in suggested:
+ self.sync_index(idx)
if ob:
self._busy = ob
- self.conn.write('receive-objects-v2\n')
+ self.conn.write('%s\n' % ob)
+ return idx
def new_packwriter(self):
self.check_busy()
self.conn.write('receive-objects-v2\n')
return PackWriter_Remote(self.conn,
objcache_maker = self._make_objcache,
- suggest_pack = self._suggest_pack,
+ suggest_packs = self._suggest_packs,
onopen = _set_busy,
onclose = self._not_busy,
ensure_busy = self.ensure_busy)
class PackWriter_Remote(git.PackWriter):
- def __init__(self, conn, objcache_maker, suggest_pack,
+ def __init__(self, conn, objcache_maker, suggest_packs,
onopen, onclose,
ensure_busy):
git.PackWriter.__init__(self, objcache_maker)
self.file = conn
self.filename = 'remote socket'
- self.suggest_pack = suggest_pack
+ self.suggest_packs = suggest_packs
self.onopen = onopen
self.onclose = onclose
self.ensure_busy = ensure_busy
def _open(self):
if not self._packopen:
- if self.onopen:
- self.onopen()
+ self.onopen()
self._packopen = True
def _end(self):
if self._packopen and self.file:
self.file.write('\0\0\0\0')
self._packopen = False
- while True:
- line = self.file.readline().strip()
- if line.startswith('index '):
- pass
- else:
- break
- id = line
- self.file.check_ok()
+ self.onclose() # Unbusy
self.objcache = None
- if self.onclose:
- self.onclose()
- if id and self.suggest_pack:
- self.suggest_pack(id)
- return id
+ return self.suggest_packs() # Returns last idx received
def close(self):
id = self._end()
assert(self.file)
if not self._packopen:
self._open()
- if self.ensure_busy:
- self.ensure_busy()
+ self.ensure_busy()
data = ''.join(datalist)
assert(data)
assert(sha)
self.count += 1
if self.file.has_input():
- line = self.file.readline().strip()
- assert(line.startswith('index '))
- idxname = line[6:]
- if self.suggest_pack:
- self.suggest_pack(idxname)
- self.objcache.refresh()
+ self.suggest_packs()
+ self.objcache.refresh()
return sha, crc
s1 = randbytes(10000)
s2 = randbytes(10000)
+s3 = randbytes(10000)
@wvtest
def test_server_split_with_indexes():
rw.new_blob(s1)
+@wvtest
+def test_multiple_suggestions():
+ os.environ['BUP_MAIN_EXE'] = '../../../bup'
+ os.environ['BUP_DIR'] = bupdir = 'buptest_tclient.tmp'
+ subprocess.call(['rm', '-rf', bupdir])
+ git.init_repo(bupdir)
+
+ lw = git.PackWriter()
+ lw.new_blob(s1)
+ lw.close()
+ lw = git.PackWriter()
+ lw.new_blob(s2)
+ lw.close()
+ WVPASSEQ(len(os.listdir(git.repo('objects/pack'))), 4)
+
+ c = client.Client(bupdir, create=True)
+ WVPASSEQ(len(os.listdir(c.cachedir)), 0)
+ rw = c.new_packwriter()
+ rw.new_blob(s1)
+ rw.new_blob(s2)
+ rw.new_blob(s3)
+ WVPASSEQ(len(os.listdir(c.cachedir)), 2)
+ rw.close()
+ WVPASSEQ(len(os.listdir(c.cachedir)), 3)
+
+
@wvtest
def test_dumb_client_server():
os.environ['BUP_MAIN_EXE'] = '../../../bup'
projects / packages / b / bup.git / commitdiff