lib/bup/cmd/server.py

   1
   2 from __future__ import absolute_import
   3 from binascii import hexlify, unhexlify
   4 import os, struct, subprocess, sys
   5
   6 from bup import options, git, vfs, vint
   7 from bup.compat import environ, hexstr, pending_raise
   8 from bup.helpers \
   9     import (Conn, debug1, debug2, finalized, linereader, lines_until_sentinel,
  10             log)
  11 from bup.io import byte_stream, path_msg
  12 from bup.repo import LocalRepo
  13
  14
  15 suspended_w = None
  16 dumb_server_mode = False
  17 repo = None
  18
  19
  20 def do_help(conn, junk):
  21     conn.write(b'Commands:\n    %s\n' % b'\n    '.join(sorted(commands)))
  22     conn.ok()
  23
  24
  25 def _set_mode():
  26     global dumb_server_mode
  27     dumb_server_mode = os.path.exists(git.repo(b'bup-dumb-server'))
  28     debug1('bup server: serving in %s mode\n'
  29            % (dumb_server_mode and 'dumb' or 'smart'))
  30
  31
  32 def _init_session(reinit_with_new_repopath=None):
  33     global repo
  34     if reinit_with_new_repopath is None and git.repodir:
  35         if not repo:
  36             repo = LocalRepo()
  37         return
  38     git.check_repo_or_die(reinit_with_new_repopath)
  39     if repo:
  40         repo.close()
  41     repo = LocalRepo()
  42     # OK. we now know the path is a proper repository. Record this path in the
  43     # environment so that subprocesses inherit it and know where to operate.
  44     environ[b'BUP_DIR'] = git.repodir
  45     debug1('bup server: bupdir is %s\n' % path_msg(git.repodir))
  46     _set_mode()
  47
  48
  49 def init_dir(conn, arg):
  50     git.init_repo(arg)
  51     debug1('bup server: bupdir initialized: %s\n' % path_msg(git.repodir))
  52     _init_session(arg)
  53     conn.ok()
  54
  55
  56 def set_dir(conn, arg):
  57     _init_session(arg)
  58     conn.ok()
  59
  60
  61 def list_indexes(conn, junk):
  62     _init_session()
  63     suffix = b''
  64     if dumb_server_mode:
  65         suffix = b' load'
  66     for f in os.listdir(git.repo(b'objects/pack')):
  67         if f.endswith(b'.idx'):
  68             conn.write(b'%s%s\n' % (f, suffix))
  69     conn.ok()
  70
  71
  72 def send_index(conn, name):
  73     _init_session()
  74     assert name.find(b'/') < 0
  75     assert name.endswith(b'.idx')
  76     with git.open_idx(git.repo(b'objects/pack/%s' % name)) as idx:
  77         conn.write(struct.pack('!I', len(idx.map)))
  78         conn.write(idx.map)
  79     conn.ok()
  80
  81
  82 def receive_objects_v2(conn, junk):
  83     global suspended_w
  84     _init_session()
  85     if suspended_w:
  86         w = suspended_w
  87         suspended_w = None
  88     elif dumb_server_mode:
  89         w = git.PackWriter(objcache_maker=None)
  90     else:
  91         w = git.PackWriter()
  92     try:
  93         suggested = set()
  94         while 1:
  95             ns = conn.read(4)
  96             if not ns:
  97                 w.abort()
  98                 raise Exception('object read: expected length header, got EOF')
  99             n = struct.unpack('!I', ns)[0]
 100             #debug2('expecting %d bytes\n' % n)
 101             if not n:
 102                 debug1('bup server: received %d object%s.\n'
 103                     % (w.count, w.count!=1 and "s" or ''))
 104                 fullpath = w.close(run_midx=not dumb_server_mode)
 105                 w = None
 106                 if fullpath:
 107                     dir, name = os.path.split(fullpath)
 108                     conn.write(b'%s.idx\n' % name)
 109                 conn.ok()
 110                 return
 111             elif n == 0xffffffff:
 112                 debug2('bup server: receive-objects suspending\n')
 113                 conn.ok()
 114                 suspended_w = w
 115                 w = None
 116                 return
 117
 118             shar = conn.read(20)
 119             crcr = struct.unpack('!I', conn.read(4))[0]
 120             n -= 20 + 4
 121             buf = conn.read(n)  # object sizes in bup are reasonably small
 122             #debug2('read %d bytes\n' % n)
 123             _check(w, n, len(buf), 'object read: expected %d bytes, got %d\n')
 124             if not dumb_server_mode:
 125                 oldpack = w.exists(shar, want_source=True)
 126                 if oldpack:
 127                     assert(not oldpack == True)
 128                     assert(oldpack.endswith(b'.idx'))
 129                     (dir,name) = os.path.split(oldpack)
 130                     if not (name in suggested):
 131                         debug1("bup server: suggesting index %s\n"
 132                                % git.shorten_hash(name).decode('ascii'))
 133                         debug1("bup server:   because of object %s\n"
 134                                % hexstr(shar))
 135                         conn.write(b'index %s\n' % name)
 136                         suggested.add(name)
 137                     continue
 138             nw, crc = w._raw_write((buf,), sha=shar)
 139             _check(w, crcr, crc, 'object read: expected crc %d, got %d\n')
 140     # py2: this clause is unneeded with py3
 141     except BaseException as ex:
 142         with pending_raise(ex):
 143             if w:
 144                 w, w_tmp = None, w
 145                 w_tmp.close()
 146     finally:
 147         if w: w.close()
 148     assert False  # should be unreachable
 149
 150
 151 def _check(w, expected, actual, msg):
 152     if expected != actual:
 153         w.abort()
 154         raise Exception(msg % (expected, actual))
 155
 156
 157 def read_ref(conn, refname):
 158     _init_session()
 159     r = git.read_ref(refname)
 160     conn.write(b'%s\n' % hexlify(r) if r else b'')
 161     conn.ok()
 162
 163
 164 def update_ref(conn, refname):
 165     _init_session()
 166     newval = conn.readline().strip()
 167     oldval = conn.readline().strip()
 168     git.update_ref(refname, unhexlify(newval), unhexlify(oldval))
 169     conn.ok()
 170
 171 def join(conn, id):
 172     _init_session()
 173     try:
 174         for blob in git.cp().join(id):
 175             conn.write(struct.pack('!I', len(blob)))
 176             conn.write(blob)
 177     except KeyError as e:
 178         log('server: error: %s\n' % e)
 179         conn.write(b'\0\0\0\0')
 180         conn.error(e)
 181     else:
 182         conn.write(b'\0\0\0\0')
 183         conn.ok()
 184
 185 def cat_batch(conn, dummy):
 186     _init_session()
 187     cat_pipe = git.cp()
 188     # For now, avoid potential deadlock by just reading them all
 189     for ref in tuple(lines_until_sentinel(conn, b'\n', Exception)):
 190         ref = ref[:-1]
 191         it = cat_pipe.get(ref)
 192         info = next(it)
 193         if not info[0]:
 194             conn.write(b'missing\n')
 195             continue
 196         conn.write(b'%s %s %d\n' % info)
 197         for buf in it:
 198             conn.write(buf)
 199     conn.ok()
 200
 201 def refs(conn, args):
 202     limit_to_heads, limit_to_tags = args.split()
 203     assert limit_to_heads in (b'0', b'1')
 204     assert limit_to_tags in (b'0', b'1')
 205     limit_to_heads = int(limit_to_heads)
 206     limit_to_tags = int(limit_to_tags)
 207     _init_session()
 208     patterns = tuple(x[:-1] for x in lines_until_sentinel(conn, b'\n', Exception))
 209     for name, oid in git.list_refs(patterns=patterns,
 210                                    limit_to_heads=limit_to_heads,
 211                                    limit_to_tags=limit_to_tags):
 212         assert b'\n' not in name
 213         conn.write(b'%s %s\n' % (hexlify(oid), name))
 214     conn.write(b'\n')
 215     conn.ok()
 216
 217 def rev_list(conn, _):
 218     _init_session()
 219     count = conn.readline()
 220     if not count:
 221         raise Exception('Unexpected EOF while reading rev-list count')
 222     assert count == b'\n'
 223     count = None
 224     fmt = conn.readline()
 225     if not fmt:
 226         raise Exception('Unexpected EOF while reading rev-list format')
 227     fmt = None if fmt == b'\n' else fmt[:-1]
 228     refs = tuple(x[:-1] for x in lines_until_sentinel(conn, b'\n', Exception))
 229     args = git.rev_list_invocation(refs, format=fmt)
 230     p = subprocess.Popen(args, env=git._gitenv(git.repodir),
 231                          stdout=subprocess.PIPE)
 232     while True:
 233         out = p.stdout.read(64 * 1024)
 234         if not out:
 235             break
 236         conn.write(out)
 237     conn.write(b'\n')
 238     rv = p.wait()  # not fatal
 239     if rv:
 240         msg = 'git rev-list returned error %d' % rv
 241         conn.error(msg)
 242         raise git.GitError(msg)
 243     conn.ok()
 244
 245 def resolve(conn, args):
 246     _init_session()
 247     (flags,) = args.split()
 248     flags = int(flags)
 249     want_meta = bool(flags & 1)
 250     follow = bool(flags & 2)
 251     have_parent = bool(flags & 4)
 252     parent = vfs.read_resolution(conn) if have_parent else None
 253     path = vint.read_bvec(conn)
 254     if not len(path):
 255         raise Exception('Empty resolve path')
 256     try:
 257         res = list(vfs.resolve(repo, path, parent=parent, want_meta=want_meta,
 258                                follow=follow))
 259     except vfs.IOError as ex:
 260         res = ex
 261     if isinstance(res, vfs.IOError):
 262         conn.write(b'\x00')  # error
 263         vfs.write_ioerror(conn, res)
 264     else:
 265         conn.write(b'\x01')  # success
 266         vfs.write_resolution(conn, res)
 267     conn.ok()
 268
 269 optspec = """
 270 bup server
 271 """
 272
 273 commands = {
 274     b'quit': None,
 275     b'help': do_help,
 276     b'init-dir': init_dir,
 277     b'set-dir': set_dir,
 278     b'list-indexes': list_indexes,
 279     b'send-index': send_index,
 280     b'receive-objects-v2': receive_objects_v2,
 281     b'read-ref': read_ref,
 282     b'update-ref': update_ref,
 283     b'join': join,
 284     b'cat': join,  # apocryphal alias
 285     b'cat-batch' : cat_batch,
 286     b'refs': refs,
 287     b'rev-list': rev_list,
 288     b'resolve': resolve
 289 }
 290
 291 def main(argv):
 292     global repo, suspended_w
 293
 294     o = options.Options(optspec)
 295     opt, flags, extra = o.parse_bytes(argv[1:])
 296     if extra:
 297         o.fatal('no arguments expected')
 298
 299     debug2('bup server: reading from stdin.\n')
 300
 301     # FIXME: this protocol is totally lame and not at all future-proof.
 302     # (Especially since we abort completely as soon as *anything* bad happens)
 303     sys.stdout.flush()
 304     with Conn(byte_stream(sys.stdin), byte_stream(sys.stdout)) as conn, \
 305          finalized(None, lambda _: repo and repo.close()), \
 306          finalized(None, lambda _: suspended_w and suspended_w.close()):
 307         lr = linereader(conn)
 308         for _line in lr:
 309             line = _line.strip()
 310             if not line:
 311                 continue
 312             debug1('bup server: command: %r\n' % line)
 313             words = line.split(b' ', 1)
 314             cmd = words[0]
 315             rest = len(words)>1 and words[1] or b''
 316             if cmd == b'quit':
 317                 break
 318             else:
 319                 cmd = commands.get(cmd)
 320                 if cmd:
 321                     cmd(conn, rest)
 322                 else:
 323                     raise Exception('unknown server command: %r\n' % line)
 324     debug1('bup server: done\n')