prep for flex file layout server

index 65e4f55..ade55e1 100644
@@ -54,6 +54,7 @@ class DataServer(object):
         # as wide as the client/MDS channel (at least for linux client)
         fore_attrs = channel_attrs4(0, 16384, 16384, 2868, 8, 8, [])
         self.sess = c.create_session(fore_attrs=fore_attrs)
+        sess.compound([op.reclaim_complete(FALSE)])
         self.make_root()
 
     def disconnect(self):

index ade55e1..aca07ee 100644
@@ -9,6 +9,7 @@ import nfs4client
 import hashlib
 import sys
 import nfs4_ops as op
+import socket
 
 log = logging.getLogger("Dataserver Manager")
 

@@ -198,7 +199,7 @@ class DSDevice(object):
                     ds = DataServer(server, port, path, mdsds=self.mdsds,
                                     multipath_servers=server_list)
                     self.list.append(ds)
-                except:
+                except socket.error:
                     log.critical("cannot access %s:%i/%s" %
                                  (server, port, '/'.join(path)))
                     sys.exit(1)

index aca07ee..435ddd2 100644
@@ -59,7 +59,7 @@ class DataServer(object):
         self.make_root()
 
     def disconnect(self):
-        self.sess.destroy()
+        pass
 
     def execute(self, ops, exceptions=[], delay=5, maxretries=3):
         """ execute the NFS call

index 2201550..f37bd9b 100644
@@ -1,8 +1,8 @@
 from __future__ import with_statement
-from pnfs_block_pack import PNFS_BLOCKPacker as Packer
-from pnfs_block_pack import PNFS_BLOCKUnpacker as Unpacker
-from pnfs_block_type import *
-from pnfs_block_const import *
+from xdrdef.pnfs_block_pack import PNFS_BLOCKPacker as Packer
+from xdrdef.pnfs_block_pack import PNFS_BLOCKUnpacker as Unpacker
+from xdrdef.pnfs_block_type import *
+from xdrdef.pnfs_block_const import *
 
 import fs_base
 from threading import Lock

index 84b7d38..225197f 100644
@@ -10,8 +10,8 @@
 
 import time
 import testmod
-from nfs4_const import *
-from nfs4_type import *
+from xdrdef.nfs4_const import *
+from xdrdef.nfs4_type import *
 import rpc
 import nfs4client
 import os

index 2238aee..4747f73 100644
@@ -1,6 +1,6 @@
-from nfs4_type import server_owner4, nfs_impl_id4
-from nfs4_const import *
-import nfs4_const
+from xdrdef.nfs4_type import server_owner4, nfs_impl_id4
+from xdrdef.nfs4_const import *
+import xdrdef.nfs4_const
 import nfs4lib
 from copy import deepcopy
 import os

@@ -37,7 +37,7 @@ def _statcode(value):
     try:
         return int(value)
     except ValueError:
-        rv = getattr(nfs4_const, value, None)
+        rv = getattr(xdr.nfs4_const, value, None)
         if rv is None or nfsstat4.get(rv, None) != value:
             raise
         else:

index 435ddd2..acca210 100644
@@ -1,8 +1,8 @@
 import rpc
 import nfs4lib
-from nfs4_type import *
-from nfs4_pack import NFS4Packer
-from nfs4_const import *
+from xdrdef.nfs4_type import *
+from xdrdef.nfs4_pack import NFS4Packer
+from xdrdef.nfs4_const import *
 import time
 import logging
 import nfs4client

index a63c5bf..3d63ce5 100644
@@ -1,6 +1,6 @@
 from nfs4state import FileState
-from nfs4_const import *
-from nfs4_type import fsid4, layout4, layout_content4, nfsv4_1_file_layout4
+from xdrdef.nfs4_const import *
+from xdrdef.nfs4_type import fsid4, layout4, layout_content4, nfsv4_1_file_layout4
 import nfs4lib
 from nfs4lib import NFS4Error
 import struct

@@ -8,7 +8,7 @@ import logging
 from locking import Lock, RWLock
 from cStringIO import StringIO
 import time
-from nfs4_pack import NFS4Packer
+from xdrdef.nfs4_pack import NFS4Packer
 
 log_o = logging.getLogger("fs.obj")
 log_fs = logging.getLogger("fs")

@@ -1070,7 +1070,7 @@ class StubFS_Disk(FileSystem):
 
 ###################################################
 
-from pnfs_block_type import pnfs_block_extent4, pnfs_block_layout4
+from xdrdef.pnfs_block_type import pnfs_block_extent4, pnfs_block_layout4
 import block
 
 class my_ro_extent(object):

index 75381b6..35a10ca 100644
@@ -2,8 +2,8 @@
 <name>() that returns the appropriate *_argop4 structure, hiding
 this routine packing from the user.
 """
-import nfs4_type as _type
-import nfs4_const as _const
+import xdrdef.nfs4_type as _type
+import xdrdef.nfs4_const as _const
 
 # This string is our general function template
 code = """\

index e750728..2aa0a2c 100644
@@ -2,10 +2,9 @@ import use_local # HACK so don't have to rebuild constantly
 import rpc
 import nfs4lib
 from nfs4lib import NFS4Error, NFS4Replay, inc_u32
-import nfs4_type, nfs4_const
-from nfs4_type import *
-from nfs4_const import *
-from sctrl_pack import SCTRLPacker, SCTRLUnpacker
+from xdrdef.nfs4_type import *
+from xdrdef.nfs4_const import *
+from xdrdef.sctrl_pack import SCTRLPacker, SCTRLUnpacker
 import nfs4_ops as op
 import time, struct
 import threading

@@ -545,8 +544,8 @@ sec.init_cred(call)
 
 """ EXCHANGE_ID
 import nfs4client
-from nfs4_type import *
-from nfs4_const import *
+from xdrdef.nfs4_type import *
+from xdrdef.nfs4_const import *
 import nfs4_ops as op
 owner = client_owner4("12345678","MyClientName")
 protect = state_protect4_a(SP4_NONE)

index 317f04c..cadd237 100644
@@ -4,10 +4,10 @@
 """
 
 import nfs4lib
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 import sys
-import nfs4_type, nfs4_const
-from nfs4_type import *
+import xdrdef.nfs4_type, xdrdef.nfs4_const
+from xdrdef.nfs4_type import *
 
 _d = {"CompoundState" : "CompoundState",
       "PairedResults" : "PairedResults",

@@ -34,8 +34,8 @@ def %(encode_status)s_by_name(name, status, *args, **kwargs):
     name_l = name.lower()
     name_u = name.upper()
     try:
-        res4 = getattr(nfs4_type, name_u + "4res")(status, *args, **kwargs)
-        result = %(nfs_resop4)s(getattr(nfs4_const, "OP_" + name_u))
+        res4 = getattr(xdrdef.nfs4_type, name_u + "4res")(status, *args, **kwargs)
+        result = %(nfs_resop4)s(getattr(xdrdef.nfs4_const, "OP_" + name_u))
         setattr(result, %(mangle)s, res4)
         # STUB XXX 4.1 has messed with the naming conventions,
         #      and added prefixes to the "status" variable. Grrr.

157 lines not on this page · raw message

index 65fb9af..2ae99fe 100755
@@ -4,10 +4,10 @@ import use_local # HACK so don't have to rebuild constantly
 import nfs4lib
 from nfs4lib import inc_u32, NFS4Error, NFS4Replay
 import rpc
-from nfs4_const import *
-from nfs4_type import *
-from sctrl_pack import SCTRLPacker, SCTRLUnpacker
-import sctrl_type, sctrl_const
+from xdrdef.nfs4_const import *
+from xdrdef.nfs4_type import *
+from xdrdef.sctrl_pack import SCTRLPacker, SCTRLUnpacker
+import xdrdef.sctrl_type, xdrdef.sctrl_const
 import traceback, threading
 from locking import Lock, Counter
 import time

@@ -620,7 +620,7 @@ class NFS4Server(rpc.Server):
             return rpc.GARBAGE_ARGS, None
         log_cfg.info(repr(args))
         # Handle the given control operation
-        opname = sctrl_const.ctrl_opnum.get(args.ctrlop, 'ctrl_illegal')
+        opname = xdrdef.sctrl_const.ctrl_opnum.get(args.ctrlop, 'ctrl_illegal')
         funct = getattr(self, opname.lower(), None)
         if funct is None:
             # This shouldn't happen

@@ -633,8 +633,8 @@ class NFS4Server(rpc.Server):
             raise
         # Now pack and return the result
         p = SCTRLPacker()
-        # res = sctrl_type.CTRLres(status, sctrl_type.resdata_t(args.ctrlop))
-        res = sctrl_type.CTRLres(status, result)
+        # res = xdrdef.sctrl_type.CTRLres(status, xdrdef.sctrl_type.resdata_t(args.ctrlop))
+        res = xdrdef.sctrl_type.CTRLres(status, result)
         p.pack_CTRLres(res)
         return rpc.SUCCESS, p.get_buffer()
 

@@ -1926,22 +1926,22 @@ class NFS4Server(rpc.Server):
 
     def ctrl_reset(self, arg):
         self.recording.reset()
-        return sctrl_const.CTRLSTAT_OK, sctrl_type.resdata_t(arg.ctrlop)
+        return xdrdef.sctrl_const.CTRLSTAT_OK, xdrdef.sctrl_type.resdata_t(arg.ctrlop)
 
     def ctrl_record(self, arg):
         self.recording.set_stamp(arg.stamp)
         self.recording.on = True
-        return sctrl_const.CTRLSTAT_OK, sctrl_type.resdata_t(arg.ctrlop)
+        return xdrdef.sctrl_const.CTRLSTAT_OK, xdrdef.sctrl_type.resdata_t(arg.ctrlop)
 
     def ctrl_pause(self, arg):
         self.recording.on = False
-        return sctrl_const.CTRLSTAT_OK, sctrl_type.resdata_t(arg.ctrlop)
+        return xdrdef.sctrl_const.CTRLSTAT_OK, xdrdef.sctrl_type.resdata_t(arg.ctrlop)
 
     def ctrl_grab(self, arg):
         queue = self.recording.queues.get(arg.stamp, None)
         if queue is None:
-            return sctrl_const.CTRLSTAT_NOT_AVAIL, \
-                   sctrl_type.resdata_t(arg.ctrlop, sctrl_type.GRABres([],[]))
+            return xdrdef.sctrl_const.CTRLSTAT_NOT_AVAIL, \
+                   xdrdef.sctrl_type.resdata_t(arg.ctrlop, xdrdef.sctrl_type.GRABres([],[]))
         max = arg.number
         if max == 0:
             max = len(queue)

@@ -1949,19 +1949,19 @@ class NFS4Server(rpc.Server):
         replies = []
         for i in range(max):
             call, reply = queue.pop()
-            if arg.dir & sctrl_const.DIR_CALL:
+            if arg.dir & xdrdef.sctrl_const.DIR_CALL:
                 calls.append(call)
-            if arg.dir & sctrl_const.DIR_REPLY:
+            if arg.dir & xdrdef.sctrl_const.DIR_REPLY:
                 replies.append(reply)
         print calls
         print replies
-        grabres = sctrl_type.GRABres(calls, replies)
-        return sctrl_const.CTRLSTAT_OK, \
-               sctrl_type.resdata_t(arg.ctrlop, grab = grabres)
+        grabres = xdrdef.sctrl_type.GRABres(calls, replies)
+        return xdrdef.sctrl_const.CTRLSTAT_OK, \
+               xdrdef.sctrl_type.resdata_t(arg.ctrlop, grab = grabres)
 
     def ctrl_illegal(self, arg):
         print "ILLEGAL"
-        return sctrl_const.CTRLSTAT_ILLEGAL, sctrl_type.resdata_t(arg.ctrlop)
+        return xdrdef.sctrl_const.CTRLSTAT_ILLEGAL, xdrdef.sctrl_type.resdata_t(arg.ctrlop)
         
     def op_setclientid(self, arg, env):
         return encode_status(NFS4ERR_NOTSUPP)

index e8445e3..1294083 100644
@@ -5,9 +5,9 @@ from locking import Lock
 import struct
 import nfs4lib
 from nfs4lib import NFS4Error
-#from nfs4_type import stateid4
-from nfs4_type import *
-from nfs4_const import *
+#from xdrdef.nfs4_type import stateid4
+from xdrdef.nfs4_type import *
+from xdrdef.nfs4_const import *
 import nfs4_ops as op
 import rpc
 import logging

index b2df732..b8b05cf 100644
@@ -10,8 +10,8 @@
 
 import time
 import testmod
-from nfs4_const import *
-from nfs4_type import *
+from xdrdef.nfs4_const import *
+from xdrdef.nfs4_type import *
 import rpc
 import nfs4client
 import sys

index f62968e..432dbbf 100644
@@ -1,5 +1,5 @@
-from nfs4_const import *
-from nfs4_type import *
+from xdrdef.nfs4_const import *
+from xdrdef.nfs4_type import *
 import nfs4_ops as op
 from environment import check, fail, create_file
 from block import Packer as BlockPacker, Unpacker as BlockUnpacker, \

index c6a5ea9..0443a29 100644
@@ -1,7 +1,7 @@
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 import nfs4_ops as op
 from environment import check, fail, get_invalid_utf8strings
-from nfs4_type import *
+from xdrdef.nfs4_type import *
 from rpc import RPCAcceptError, GARBAGE_ARGS, RPCTimeout
 from nfs4lib import NFS4Error, hash_oids, encrypt_oids, FancyNFS4Packer
 

index ae5ca27..4685eb0 100644
@@ -1,7 +1,7 @@
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 import nfs4_ops as op
 from environment import check, fail, create_file, open_file
-from nfs4_type import *
+from xdrdef.nfs4_type import *
 import random
 import nfs4lib
 import threading

index 9a1adc4..b2e69c5 100644
@@ -1,11 +1,11 @@
 from st_create_session import create_session
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 
 from environment import check, checklist, fail, create_file, open_file, close_file
 from environment import open_create_file_op, use_obj
-from nfs4_type import open_owner4, openflag4, createhow4, open_claim4
-from nfs4_type import creatverfattr, fattr4, stateid4, locker4, lock_owner4
-from nfs4_type import open_to_lock_owner4
+from xdrdef.nfs4_type import open_owner4, openflag4, createhow4, open_claim4
+from xdrdef.nfs4_type import creatverfattr, fattr4, stateid4, locker4, lock_owner4
+from xdrdef.nfs4_type import open_to_lock_owner4
 import nfs4_ops as op
 import threading
 

index aff6842..7bf6b5b 100644
@@ -1,7 +1,7 @@
 from st_create_session import create_session
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, fail, create_file
-from nfs4_type import open_owner4, openflag4, createhow4, open_claim4
+from xdrdef.nfs4_type import open_owner4, openflag4, createhow4, open_claim4
 import nfs4_ops as op
 import threading
 

index 02ce719..5abc904 100644
@@ -1,9 +1,9 @@
 from st_create_session import create_session
 from st_open import open_claim4
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 
 from environment import check, checklist, fail, create_file, open_file, close_file
-from nfs4_type import *
+from xdrdef.nfs4_type import *
 import nfs4_ops as op
 import nfs4lib
 import threading

index 1b193cf..6b7042f 100644
@@ -1,5 +1,5 @@
 from st_create_session import create_session
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, fail
 import nfs4_ops as op
 import nfs4lib

index f4f0762..e3c3424 100644
@@ -1,7 +1,7 @@
 from st_create_session import create_session
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, checklist, fail, create_file, open_file
-from nfs4_type import open_owner4, openflag4, createhow4, open_claim4
+from xdrdef.nfs4_type import open_owner4, openflag4, createhow4, open_claim4
 import nfs4_ops as op
 import threading
 import rpc

index 956090f..39b8b0c 100644
@@ -1,8 +1,8 @@
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 import nfs4_ops as op
 import time
 from environment import check, checklist, fail
-from nfs4_type import *
+from xdrdef.nfs4_type import *
 from rpc import RPCAcceptError, GARBAGE_ARGS, RPCTimeout
 from nfs4lib import NFS4Error, hash_oids, encrypt_oids
 

index 7a10c6c..df0107d 100644
@@ -1,5 +1,5 @@
-from nfs4_const import *
-from nfs4_type import *
+from xdrdef.nfs4_const import *
+from xdrdef.nfs4_type import *
 from environment import check, fail, use_obj, open_file, create_file
 import nfs4_ops as op
 from block import Packer as BlockPacker, Unpacker as BlockUnpacker, \

index 33e06b9..85ddcfe 100644
@@ -1,4 +1,4 @@
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, fail
 import nfs4_ops as op
 

index 512ea6b..7aa7406 100644
@@ -1,4 +1,4 @@
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, fail, use_obj
 import nfs4_ops as op
 

index 2ec646f..71e01fb 100644
@@ -1,11 +1,11 @@
 from st_create_session import create_session
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 
 from environment import check, checklist, fail, create_file, open_file, close_file
 from environment import open_create_file_op
-from nfs4_type import open_owner4, openflag4, createhow4, open_claim4
-from nfs4_type import creatverfattr, fattr4, stateid4, locker4, lock_owner4
-from nfs4_type import open_to_lock_owner4
+from xdrdef.nfs4_type import open_owner4, openflag4, createhow4, open_claim4
+from xdrdef.nfs4_type import creatverfattr, fattr4, stateid4, locker4, lock_owner4
+from xdrdef.nfs4_type import open_to_lock_owner4
 import nfs4_ops as op
 import threading
 

index d7bc0be..6d96048 100644
@@ -1,4 +1,4 @@
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, fail, use_obj, create_confirm, close_file
 import nfs4_ops as op
 

index bbfc40e..b778fd7 100644
@@ -1,5 +1,5 @@
-from nfs4_const import *
-from nfs4_type import *
+from xdrdef.nfs4_const import *
+from xdrdef.nfs4_type import *
 from environment import check, checklist, fail, create_file, open_file, create_confirm
 import sys
 import os

index f6baf09..1d9270f 100644
@@ -1,5 +1,5 @@
 from st_create_session import create_session
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, fail, open_file, create_file, create_confirm
 import nfs4_ops as op
 import nfs4lib

index 943587c..51b6cd6 100644
@@ -1,7 +1,7 @@
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, checklist, fail, maketree, rename_obj, get_invalid_utf8strings, create_obj, create_confirm, link, use_obj, create_file
 import nfs4_ops as op
-from nfs4_type import *
+from xdrdef.nfs4_type import *
 
 def testValidDir(t, env):
     """RENAME : normal operation

index d9a1e98..b676edc 100644
@@ -1,7 +1,7 @@
 from st_create_session import create_session
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, fail, use_obj, bad_sessionid, create_file
-from nfs4_type import channel_attrs4
+from xdrdef.nfs4_type import channel_attrs4
 import nfs4_ops as op
 import nfs4lib
 

index ada4853..2f8ee1b 100644
@@ -1,7 +1,7 @@
 from st_create_session import create_session
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, fail, bad_sessionid, create_file
-from nfs4_type import channel_attrs4
+from xdrdef.nfs4_type import channel_attrs4
 import nfs4_ops as op
 import nfs4lib
 

index 96f5e42..e21bf7d 100644
@@ -1,7 +1,7 @@
 from st_create_session import create_session
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 from environment import check, fail, bad_sessionid, create_file
-from nfs4_type import channel_attrs4
+from xdrdef.nfs4_type import channel_attrs4
 import nfs4_ops as op
 import nfs4lib
 

index b984425..142b924 100644
@@ -1,7 +1,7 @@
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 import nfs4_ops as op
 from environment import check, fail
-from nfs4_type import *
+from xdrdef.nfs4_type import *
 import random
 import nfs4lib
 import threading

index a1bc5ca..cfb3809 100644
@@ -1,4 +1,4 @@
-from nfs4_const import *
+from xdrdef.nfs4_const import *
 import nfs4_ops as op
 from environment import check, checklist, get_invalid_clientid, makeStaleId, \
     do_getattrdict, use_obj

index e73a1d0..5c3a0df 100644
@@ -31,22 +31,25 @@ class build_py(_build_py):
                 self.build_module(module, module_file, package)
 
     def expand_xdr(self, dir):
+        print "expand = %r" % dir
         cwd = os.getcwd()
-        try:
-            if dir:
-                os.chdir(dir)
-            xdr_files = glob(os.path.join(dir, "*.x"))
-            for f in xdr_files:
-                # Can conditionalize this
-                # XXX need some way to pass options here
-                xdrgen.run(f)
-                try:
-                    os.remove("parser.out")
-                    os.remove("parsetab.py")
-                except:
-                    print "Remove parse* failed"
-        finally:
-            os.chdir(cwd)
+        xdrdir = os.path.join(cwd, dir, 'xdrdef')
+        print "xdrdir = %r" % xdrdir
+        if os.path.exists(xdrdir):
+            try:
+                os.chdir(xdrdir)
+                xdr_files = glob(os.path.join(xdrdir, "*.x"))
+                for f in xdr_files:
+                    # Can conditionalize this
+                    # XXX need some way to pass options here
+                    xdrgen.run(f)
+                    try:
+                        os.remove("parser.out")
+                        os.remove("parsetab.py")
+                    except:
+                        print "Remove parse* failed"
+            finally:
+                os.chdir(cwd)
 
 setup(name = "nfs4",
       version = "0.0.0", # import this?

index 9189c60..95e90e1 100755
@@ -193,7 +193,7 @@ def run_filter(test, options):
 
 def printflags(list):
     """Print all legal flag names, which are given in list"""
-    from nfs4_const import nfs_opnum4
+    from xdrdef.nfs4_const import nfs_opnum4
     command_names = [s.lower()[3:].replace('_', '') \
                      for s in nfs_opnum4.values()]
     list.sort()

index 61e308c..7c10fe6 100755
@@ -190,7 +190,7 @@ def run_filter(test, options):
 
 def printflags(list):
     """Print all legal flag names, which are given in list"""
-    from nfs4_const import nfs_opnum4
+    from xdrdef.nfs4_const import nfs_opnum4
     command_names = [s.lower()[3:].replace('_', '') \
                      for s in nfs_opnum4.values()]
     list.sort()

new file mode 100644
index 0000000..e69de29

similarity index 100%
rename from nfs4.1/nfs4.x
rename to nfs4.1/xdrdef/nfs4.x

similarity index 100%
rename from nfs4.1/nfs4.x.diff
rename to nfs4.1/xdrdef/nfs4.x.diff

similarity index 100%
rename from nfs4.1/pnfs_block.x
rename to nfs4.1/xdrdef/pnfs_block.x

similarity index 100%
rename from nfs4.1/sctrl.x
rename to nfs4.1/xdrdef/sctrl.x

index 2aa0a2c..7a07ab7 100644
@@ -8,15 +8,13 @@ from xdrdef.sctrl_pack import SCTRLPacker, SCTRLUnpacker
 import nfs4_ops as op
 import time, struct
 import threading
-import collections
 import hmac
-from locking import Lock
 from nfs4commoncode import CBCompoundState as CompoundState, \
      cb_encode_status as encode_status, \
      cb_encode_status_by_name as encode_status_by_name
 
 
-import sys, traceback
+import traceback
 import logging
 logging.basicConfig(level=logging.INFO,
                     format="%(levelname)-7s:%(name)s:%(message)s")

index acca210..c400d22 100644
@@ -182,7 +182,7 @@ class DSDevice(object):
                 line = line.strip()
                 if not line or line.startswith('#'):
                     continue
-                print "Analyzing: %r" % line
+                log.info("Analyzing: %r" % line)
                 try:
                     server_list, path = nfs4lib.parse_nfs_url(line)
                 except:

@@ -192,7 +192,6 @@ class DSDevice(object):
                 # for now, just use the last path for local connections
                 server, port = server_list[-1]
                 server_list = server_list[:-1]
-                print server, port, path
                 try:
                     log.info("Adding dataserver ip:%s port:%s path:%s" %
                              (server, port, '/'.join(path)))

index 3d63ce5..6ef283b 100644
@@ -13,7 +13,6 @@ from xdrdef.nfs4_pack import NFS4Packer
 log_o = logging.getLogger("fs.obj")
 log_fs = logging.getLogger("fs")
 logging.addLevelName(5, "FUNCT")
-log_fs.setLevel(20)
 
 class MetaData(object):
     """Contains everything that needs to be stored

@@ -743,7 +742,6 @@ class ConfigObj(FSObject):
             line = line.strip()
             if line and not line.startswith("#"):
                 lines.append(line)
-        print lines
         if len(lines) != 1:
             self._reset()
             return

@@ -1154,7 +1152,6 @@ class LayoutFSObj(FSObject):
                                             disk_offset,
                                             e.state))
         block_layout = pnfs_block_layout4(elist)
-        print block_layout
         p = block.Packer()
         p.pack_pnfs_block_layout4(block_layout)
 ##         if self.id <= 4:

@@ -1194,7 +1191,6 @@ class LayoutFSObj(FSObject):
                 log_o.exception("Problem decoding opaque")
                 raise NFS4Error(NFS4ERR_BADLAYOUT, tag="Error decoding opaque")
             upd_list = update.blu_commit_list
-            print upd_list
         # Error check
         for e in upd_list:
             if e.bex_state != block.PNFS_BLOCK_READWRITE_DATA:

index 7a07ab7..0ba84bd 100644
@@ -19,9 +19,7 @@ import logging
 logging.basicConfig(level=logging.INFO,
                     format="%(levelname)-7s:%(name)s:%(message)s")
 log_cb = logging.getLogger("nfs.client.cb")
-log_cb.setLevel(logging.DEBUG)
 
-SHOW_TRAFFIC = True # Debugging aid, prints out client traffic
 class NFS4Client(rpc.Client, rpc.Server):
     def __init__(self, host='localhost', port=2049, minorversion=1, ctrl_proc=16):
         rpc.Client.__init__(self, 100003, 4)

@@ -76,9 +74,7 @@ class NFS4Client(rpc.Client, rpc.Server):
             pipe = self.c1
         p = packer(check_enum=checks, check_array=checks)
         c4 = COMPOUND4args(tag, version, ops)
-        if SHOW_TRAFFIC:
-            print
-            print c4
+        log_cb.info("compound args = %r" % (c4,))
         p.pack_COMPOUND4args(c4)
         return self.send_call(pipe, 1, p.get_buffer(), credinfo)
 

@@ -86,19 +82,16 @@ class NFS4Client(rpc.Client, rpc.Server):
         xid = self.compound_async(*args, **kwargs)
         pipe = kwargs.get("pipe", None)
         res = self.listen(xid, pipe=pipe)
-        if SHOW_TRAFFIC:
-            print res
+        log_cb.info("compound result = %r" % (res,))
         return res
     
     def listen(self, xid, pipe=None, timeout=10.0):
         if pipe is None:
             pipe = self.c1
         header, data = pipe.listen(xid, timeout)
-        #print "HEADER", header
         if data:
             p = nfs4lib.FancyNFS4Unpacker(data)
             data = p.unpack_COMPOUND4res()
-        #print "DATA", repr(data)
         return data
 
     def handle_0(self, data, cred):

150 lines not on this page · raw message

index 1294083..2f3cd59 100644
@@ -13,7 +13,6 @@ import rpc
 import logging
 
 log = logging.getLogger("nfs.server.state")
-log.setLevel(10)
 
 POSIXLOCK = False
 

index c400d22..dd81b20 100644
@@ -13,8 +13,8 @@ import socket
 
 log = logging.getLogger("Dataserver Manager")
 
-class DataServer(object):
-    def __init__(self, server, port, path, flavor=rpc.AUTH_SYS, active=True, mdsds=True, multipath_servers=None):
+class DataServer41(object):
+    def __init__(self, server, port, path, flavor=rpc.AUTH_SYS, active=True, mdsds=True, multipath_servers=None, summary=None):
         self.mdsds = mdsds
         self.server = server
         self.port = int(port)

@@ -32,6 +32,8 @@ class DataServer(object):
         else:
             self.multipath_servers = []
 
+        self.summary = summary
+
         if active:
             self.up()
 

@@ -47,7 +49,8 @@ class DataServer(object):
         # only support root with AUTH_SYS for now
         s1 = rpc.security.instance(rpc.AUTH_SYS)
         self.cred1 = s1.init_cred(uid=0, gid=0)
-        self.c1 = nfs4client.NFS4Client(self.server, self.port)
+        self.c1 = nfs4client.NFS4Client(self.server, self.port,
+                                        summary=self.summary)
         self.c1.set_cred(self.cred1)
         self.c1.null()
         c = self.c1.new_client("DS.init_%s" % self.server)

@@ -173,7 +176,7 @@ class DSDevice(object):
         self.address_body = None # set by load()
         self.mdsds = mdsds # if you are both the DS and the MDS we are the only server
 
-    def load(self, filename):
+    def load(self, filename, server_obj):
         """ Read dataservers from configuration file:
         where each line has format e.g. server[:[port][/path]]
         """

@@ -195,8 +198,9 @@ class DSDevice(object):
                 try:
                     log.info("Adding dataserver ip:%s port:%s path:%s" %
                              (server, port, '/'.join(path)))
-                    ds = DataServer(server, port, path, mdsds=self.mdsds,
-                                    multipath_servers=server_list)
+                    ds = DataServer41(server, port, path, mdsds=self.mdsds,
+                                    multipath_servers=server_list,
+                                    summary=server_obj.summary)
                     self.list.append(ds)
                 except socket.error:
                     log.critical("cannot access %s:%i/%s" %

index 0ba84bd..4ae884a 100644
@@ -21,7 +21,7 @@ logging.basicConfig(level=logging.INFO,
 log_cb = logging.getLogger("nfs.client.cb")
 
 class NFS4Client(rpc.Client, rpc.Server):
-    def __init__(self, host='localhost', port=2049, minorversion=1, ctrl_proc=16):
+    def __init__(self, host='localhost', port=2049, minorversion=1, ctrl_proc=16, summary=None):
         rpc.Client.__init__(self, 100003, 4)
         self.prog = 0x40000000
         self.versions = [1] # List of supported versions of prog

@@ -36,6 +36,7 @@ class NFS4Client(rpc.Client, rpc.Server):
         self.c1 = self.connect(self.server_address)
         self.sessions = {} # XXX Really, this should be per server
         self.ctrl_proc = ctrl_proc
+        self.summary = summary
 
     def set_cred(self, credinfo):
         self.default_cred = credinfo

@@ -83,6 +84,10 @@ class NFS4Client(rpc.Client, rpc.Server):
         pipe = kwargs.get("pipe", None)
         res = self.listen(xid, pipe=pipe)
         log_cb.info("compound result = %r" % (res,))
+        if self.summary:
+            self.summary.show_op('call v4.1 %s:%s' % self.server_address,
+                [ nfs_opnum4[a.argop].lower()[3:] for a in args[0] ],
+                nfsstat4[res.status])
         return res
     
     def listen(self, xid, pipe=None, timeout=10.0):

87 lines not on this page · raw message

index d96b27b..ef857ee 100644
@@ -15,7 +15,7 @@ def mount_stuff(server, opts):
         E = BlockLayoutFS(5, backing_device=dev)
         server.mount(E, path="/block")
     if opts.use_files:
-        dservers = _load_dataservers(opts.dataservers, server.is_ds and server.is_mds)
+        dservers = _load_dataservers(opts.dataservers, server)
         if dservers is None:
             return
         F = FileLayoutFS(6, dservers)

@@ -33,7 +33,8 @@ def _create_simple_block_dev():
     c1 = Concat([s3, s1])
     return BlockVolume(c1)
 
-def _load_dataservers(file, connect_to_ds=False):
+def _load_dataservers(filename, server):
+    connect_to_ds = server.is_ds and server.is_mds
     dss = DSDevice(connect_to_ds)
-    dss.load(file)
+    dss.load(filename, server)
     return dss;

index dd81b20..16be4c4 100644
@@ -64,7 +64,7 @@ class DataServer41(object):
     def disconnect(self):
         pass
 
-    def execute(self, ops, exceptions=[], delay=5, maxretries=3):
+    def _execute(self, ops, exceptions=[], delay=5, maxretries=3):
         """ execute the NFS call
         If an error code is specified in the exceptions it means that the
         caller wants to handle the error himself

@@ -118,16 +118,16 @@ class DataServer41(object):
         kind = createtype4(NF4DIR)
         for comp in self.path:
             existing_path.append(comp)
-            res = self.execute(nfs4lib.use_obj(existing_path),
+            res = self._execute(nfs4lib.use_obj(existing_path),
                                exceptions=[NFS4ERR_NOENT])
             if res.status == NFS4ERR_NOENT:
                 cr_ops = nfs4lib.use_obj(existing_path[:-1]) + \
                     [op.create(kind, comp, attrs)]
-                self.execute(cr_ops)
-        res = self.execute(nfs4lib.use_obj(self.path) + [op.getfh()])
+                self._execute(cr_ops)
+        res = self._execute(nfs4lib.use_obj(self.path) + [op.getfh()])
         self.path_fh = res.resarray[-1].object
         need = ACCESS4_READ | ACCESS4_LOOKUP | ACCESS4_MODIFY | ACCESS4_EXTEND
-        res = self.execute(nfs4lib.use_obj(self.path_fh) + [op.access(need)])
+        res = self._execute(nfs4lib.use_obj(self.path_fh) + [op.access(need)])
         if res.resarray[-1].access != need:
             raise RuntimeError
         # XXX clean DS directory

@@ -147,7 +147,7 @@ class DataServer41(object):
             open_op = op.open(seqid, access, deny,
                               open_owner4(self.sess.client.clientid, owner),
                               openflag, open_claim4(CLAIM_NULL, name))
-            res = self.execute(nfs4lib.use_obj(self.path_fh) + [open_op, op.getfh()], exceptions=[NFS4ERR_EXIST])
+            res = self._execute(nfs4lib.use_obj(self.path_fh) + [open_op, op.getfh()], exceptions=[NFS4ERR_EXIST])
             if res.status == NFS4_OK:
                  ds_fh = res.resarray[-1].opgetfh.resok4.object
                  ds_openstateid = stateid4(0, res.resarray[-2].stateid.other)

@@ -163,10 +163,37 @@ class DataServer41(object):
         seqid=0 #FIXME: seqid must be !=0
         fh, stateid = self.filehandles[mds_fh]
         ops = [op.putfh(fh)] + [op.close(seqid, stateid)]
-        res = self.execute(ops)
+        res = self._execute(ops)
         # ignoring return
         del self.filehandles[mds_fh]
 
+    def read(self, fh, pos, count):
+        ops = [op.putfh(fh),
+               op.read(nfs4lib.state00, pos, count)]
+        # There are all sorts of error handling issues here
+        res = self._execute(ops)
+        data = res.resarray[-1].data
+        return data
+
+    def write(self, fh, pos, data):
+        ops = [op.putfh(fh),
+               op.write(nfs4lib.state00, pos, FILE_SYNC4, data)]
+        # There are all sorts of error handling issues here
+        res = self._execute(ops)
+
+    def truncate(self, fh, size):
+        ops = [op.putfh(fh),
+               op.setattr(nfs4lib.state00, {FATTR4_SIZE: size})]
+        res = self._execute(ops)
+
+    def get_size(self, fh):
+        ops = [op.putfh(fh),
+               op.getattr(1L << FATTR4_SIZE)]
+        res = self._execute(ops)
+        attrdict = res.resarray[-1].obj_attributes
+        return attrdict.get(FATTR4_SIZE, 0)
+
+
 class DSDevice(object):
     def __init__(self, mdsds):
         self.list = [] # list of DataServer instances

index 6ef283b..8fc49ef 100644
@@ -1567,12 +1567,7 @@ class FilelayoutVolWrapper(object):
         self._pos = 0
 
     def read(self, count):
-        # STUB stateid0 is illegal to a ds
-        ops = [op.putfh(self._fh),
-               op.read(nfs4lib.state00, self._pos, count)]
-        # There are all sorts of error handling issues here
-        res = self._ds.execute(ops)
-        data = res.resarray[-1].data
+        data = self._ds.read(self._fh, self._pos, count)
         self._pos += len(data)
         return data
 

@@ -1580,25 +1575,14 @@ class FilelayoutVolWrapper(object):
         self._pos = offset
 
     def write(self, data):
-        ops = [op.putfh(self._fh),
-               op.write(nfs4lib.state00, self._pos, FILE_SYNC4, data)]
-        # There are all sorts of error handling issues here
-        res = self._ds.execute(ops)
+        self._ds.write(self._fh, self._pos, data)
         self._pos += len(data)
-        return
 
     def truncate(self, size):
-        ops = [op.putfh(self._fh),
-               op.setattr(nfs4lib.state00, {FATTR4_SIZE: size})]
-        res = self._ds.execute(ops)
-        return
+        self._ds.truncate(self._fh, size)
 
     def get_size(self):
-        ops = [op.putfh(self._fh),
-               op.getattr(1L << FATTR4_SIZE)]
-        res = self._ds.execute(ops)
-        attrdict = res.resarray[-1].obj_attributes
-        return attrdict.get(FATTR4_SIZE, 0)
+        return self._ds.get_size(self._fh)
 
 ################################################
 

index 16be4c4..f9e9740 100644
@@ -1,8 +1,8 @@
 import rpc
 import nfs4lib
-from xdrdef.nfs4_type import *
+import xdrdef.nfs4_type as type4
 from xdrdef.nfs4_pack import NFS4Packer
-from xdrdef.nfs4_const import *
+import xdrdef.nfs4_const as const4
 import time
 import logging
 import nfs4client

@@ -56,7 +56,7 @@ class DataServer41(object):
         c = self.c1.new_client("DS.init_%s" % self.server)
         # This is a hack to ensure MDS/DS communication path is at least
         # as wide as the client/MDS channel (at least for linux client)
-        fore_attrs = channel_attrs4(0, 16384, 16384, 2868, 8, 8, [])
+        fore_attrs = type4.channel_attrs4(0, 16384, 16384, 2868, 8, 8, [])
         self.sess = c.create_session(fore_attrs=fore_attrs)
         sess.compound([op.reclaim_complete(FALSE)])
         self.make_root()

@@ -69,12 +69,12 @@ class DataServer41(object):
         If an error code is specified in the exceptions it means that the
         caller wants to handle the error himself
         """
-        retry_errors = [NFS4ERR_DELAY, NFS4ERR_GRACE]
-        state_errors = [NFS4ERR_STALE_CLIENTID, NFS4ERR_BADSESSION,
-                        NFS4ERR_BADSLOT, NFS4ERR_DEADSESSION]
+        retry_errors = [const4.NFS4ERR_DELAY, const4.NFS4ERR_GRACE]
+        state_errors = [const4.NFS4ERR_STALE_CLIENTID, const4.NFS4ERR_BADSESSION,
+                        const4.NFS4ERR_BADSLOT, const4.NFS4ERR_DEADSESSION]
         while True:
             res = self.sess.compound(ops)
-            if res.status == NFS4_OK or res.status in exceptions:
+            if res.status == const4.NFS4_OK or res.status in exceptions:
                 return res
             elif res.status in retry_errors:
                 if maxretries > 0:

@@ -96,7 +96,7 @@ class DataServer41(object):
         uaddr = '.'.join([self.server,
                           str(self.port >> 8),
                           str(self.port & 0xff)])
-        return netaddr4(self.proto, uaddr)
+        return type4.netaddr4(self.proto, uaddr)
 
     def get_multipath_netaddr4s(self):
         netaddr4s = []

@@ -109,24 +109,24 @@ class DataServer41(object):
             if server.find(':') >= 0:
                 proto = "tcp6"
 
-            netaddr4s.append(netaddr4(proto, uaddr))
+            netaddr4s.append(type4.netaddr4(proto, uaddr))
         return netaddr4s
 
 
-    def make_root(self, attrs={FATTR4_MODE:0777}):
+    def make_root(self, attrs={const4.FATTR4_MODE:0777}):
         existing_path = []
-        kind = createtype4(NF4DIR)
+        kind = type4.createtype4(const4.NF4DIR)
         for comp in self.path:
             existing_path.append(comp)
             res = self._execute(nfs4lib.use_obj(existing_path),
-                               exceptions=[NFS4ERR_NOENT])
-            if res.status == NFS4ERR_NOENT:
+                               exceptions=[const4.NFS4ERR_NOENT])
+            if res.status == const4.NFS4ERR_NOENT:
                 cr_ops = nfs4lib.use_obj(existing_path[:-1]) + \
                     [op.create(kind, comp, attrs)]
                 self._execute(cr_ops)
         res = self._execute(nfs4lib.use_obj(self.path) + [op.getfh()])
         self.path_fh = res.resarray[-1].object
-        need = ACCESS4_READ | ACCESS4_LOOKUP | ACCESS4_MODIFY | ACCESS4_EXTEND
+        need = const4.ACCESS4_READ | const4.ACCESS4_LOOKUP | const4.ACCESS4_MODIFY | const4.ACCESS4_EXTEND
         res = self._execute(nfs4lib.use_obj(self.path_fh) + [op.access(need)])
         if res.resarray[-1].access != need:
             raise RuntimeError

@@ -136,25 +136,25 @@ class DataServer41(object):
         return hashlib.sha1("%r" % mds_fh).hexdigest()
 
     def open_file(self, mds_fh, seqid=0,
-                  access=OPEN4_SHARE_ACCESS_BOTH, deny=OPEN4_SHARE_DENY_NONE,
-                  attrs={FATTR4_MODE: 0777}, owner = "mds", mode=GUARDED4):
+                  access=const4.OPEN4_SHARE_ACCESS_BOTH, deny=const4.OPEN4_SHARE_DENY_NONE,
+                  attrs={const4.FATTR4_MODE: 0777}, owner = "mds", mode=const4.GUARDED4):
         verifier = self.sess.c.verifier
-        openflag = openflag4(OPEN4_CREATE, createhow4(mode, attrs, verifier))
+        openflag = type4.openflag4(const4.OPEN4_CREATE, type4.createhow4(mode, attrs, verifier))
         name = self.fh_to_name(mds_fh)
         while True:
             if mds_fh in self.filehandles:
                 return
             open_op = op.open(seqid, access, deny,
-                              open_owner4(self.sess.client.clientid, owner),
-                              openflag, open_claim4(CLAIM_NULL, name))
-            res = self._execute(nfs4lib.use_obj(self.path_fh) + [open_op, op.getfh()], exceptions=[NFS4ERR_EXIST])
-            if res.status == NFS4_OK:
+                              type4.open_owner4(self.sess.client.clientid, owner),
+                              openflag, type4.open_claim4(const4.CLAIM_NULL, name))
+            res = self._execute(nfs4lib.use_obj(self.path_fh) + [open_op, op.getfh()], exceptions=[const4.NFS4ERR_EXIST])
+            if res.status == const4.NFS4_OK:
                  ds_fh = res.resarray[-1].opgetfh.resok4.object
-                 ds_openstateid = stateid4(0, res.resarray[-2].stateid.other)
+                 ds_openstateid = type4.stateid4(0, res.resarray[-2].stateid.other)
                  self.filehandles[mds_fh] = (ds_fh, ds_openstateid)
                  return
-            elif res.status == NFS4ERR_EXIST:
-                 openflag = openflag4(OPEN4_NOCREATE)
+            elif res.status == const4.NFS4ERR_EXIST:
+                 openflag = type4.openflag4(const4.OPEN4_NOCREATE)
             else:
                 raise RuntimeError
 

@@ -177,26 +177,26 @@ class DataServer41(object):
 
     def write(self, fh, pos, data):
         ops = [op.putfh(fh),
-               op.write(nfs4lib.state00, pos, FILE_SYNC4, data)]
+               op.write(nfs4lib.state00, pos, const4.FILE_SYNC4, data)]
         # There are all sorts of error handling issues here
         res = self._execute(ops)
 
     def truncate(self, fh, size):
         ops = [op.putfh(fh),
-               op.setattr(nfs4lib.state00, {FATTR4_SIZE: size})]
+               op.setattr(nfs4lib.state00, {const4.FATTR4_SIZE: size})]
         res = self._execute(ops)
 
     def get_size(self, fh):
         ops = [op.putfh(fh),
-               op.getattr(1L << FATTR4_SIZE)]
+               op.getattr(1L << const4.FATTR4_SIZE)]
         res = self._execute(ops)
         attrdict = res.resarray[-1].obj_attributes
-        return attrdict.get(FATTR4_SIZE, 0)
+        return attrdict.get(const4.FATTR4_SIZE, 0)
 
 
 class DSDevice(object):
     def __init__(self, mdsds):
-        self.list = [] # list of DataServer instances
+        self.list = [] # list of DataServer41 instances
         # STUB only one data group supported for now
         self.devid = 0
         self.active = 0

@@ -248,7 +248,7 @@ class DSDevice(object):
                 netaddrs.append(multipath)
                 stripe_indices.append(index)
                 index = index + 1
-        addr = nfsv4_1_file_layout_ds_addr4(stripe_indices, netaddrs)
+        addr = type4.nfsv4_1_file_layout_ds_addr4(stripe_indices, netaddrs)
         p = NFS4Packer()
         p.pack_nfsv4_1_file_layout_ds_addr4(addr)
         return p.get_buffer()

index f9e9740..d092cf7 100644
@@ -8,11 +8,13 @@ import logging
 import nfs4client
 import hashlib
 import sys
-import nfs4_ops as op
+import nfs_ops
 import socket
 
 log = logging.getLogger("Dataserver Manager")
 
+op4 = nfs_ops.NFS4ops()
+
 class DataServer41(object):
     def __init__(self, server, port, path, flavor=rpc.AUTH_SYS, active=True, mdsds=True, multipath_servers=None, summary=None):
         self.mdsds = mdsds

@@ -122,12 +124,12 @@ class DataServer41(object):
                                exceptions=[const4.NFS4ERR_NOENT])
             if res.status == const4.NFS4ERR_NOENT:
                 cr_ops = nfs4lib.use_obj(existing_path[:-1]) + \
-                    [op.create(kind, comp, attrs)]
+                    [op4.create(kind, comp, attrs)]
                 self._execute(cr_ops)
-        res = self._execute(nfs4lib.use_obj(self.path) + [op.getfh()])
+        res = self._execute(nfs4lib.use_obj(self.path) + [op4.getfh()])
         self.path_fh = res.resarray[-1].object
         need = const4.ACCESS4_READ | const4.ACCESS4_LOOKUP | const4.ACCESS4_MODIFY | const4.ACCESS4_EXTEND
-        res = self._execute(nfs4lib.use_obj(self.path_fh) + [op.access(need)])
+        res = self._execute(nfs4lib.use_obj(self.path_fh) + [op4.access(need)])
         if res.resarray[-1].access != need:
             raise RuntimeError
         # XXX clean DS directory

@@ -144,10 +146,10 @@ class DataServer41(object):
         while True:
             if mds_fh in self.filehandles:
                 return
-            open_op = op.open(seqid, access, deny,
+            open_op = op4.open(seqid, access, deny,
                               type4.open_owner4(self.sess.client.clientid, owner),
                               openflag, type4.open_claim4(const4.CLAIM_NULL, name))
-            res = self._execute(nfs4lib.use_obj(self.path_fh) + [open_op, op.getfh()], exceptions=[const4.NFS4ERR_EXIST])
+            res = self._execute(nfs4lib.use_obj(self.path_fh) + [open_op, op4.getfh()], exceptions=[const4.NFS4ERR_EXIST])
             if res.status == const4.NFS4_OK:
                  ds_fh = res.resarray[-1].opgetfh.resok4.object
                  ds_openstateid = type4.stateid4(0, res.resarray[-2].stateid.other)

@@ -162,33 +164,33 @@ class DataServer41(object):
         """close the given file"""
         seqid=0 #FIXME: seqid must be !=0
         fh, stateid = self.filehandles[mds_fh]
-        ops = [op.putfh(fh)] + [op.close(seqid, stateid)]
+        ops = [op4.putfh(fh)] + [op4.close(seqid, stateid)]
         res = self._execute(ops)
         # ignoring return
         del self.filehandles[mds_fh]
 
     def read(self, fh, pos, count):
-        ops = [op.putfh(fh),
-               op.read(nfs4lib.state00, pos, count)]
+        ops = [op4.putfh(fh),
+               op4.read(nfs4lib.state00, pos, count)]
         # There are all sorts of error handling issues here
         res = self._execute(ops)
         data = res.resarray[-1].data
         return data
 
     def write(self, fh, pos, data):
-        ops = [op.putfh(fh),
-               op.write(nfs4lib.state00, pos, const4.FILE_SYNC4, data)]
+        ops = [op4.putfh(fh),
+               op4.write(nfs4lib.state00, pos, const4.FILE_SYNC4, data)]
         # There are all sorts of error handling issues here
         res = self._execute(ops)
 
     def truncate(self, fh, size):
-        ops = [op.putfh(fh),
-               op.setattr(nfs4lib.state00, {const4.FATTR4_SIZE: size})]
+        ops = [op4.putfh(fh),
+               op4.setattr(nfs4lib.state00, {const4.FATTR4_SIZE: size})]
         res = self._execute(ops)
 
     def get_size(self, fh):
-        ops = [op.putfh(fh),
-               op.getattr(1L << const4.FATTR4_SIZE)]
+        ops = [op4.putfh(fh),
+               op4.getattr(1L << const4.FATTR4_SIZE)]
         res = self._execute(ops)
         attrdict = res.resarray[-1].obj_attributes
         return attrdict.get(const4.FATTR4_SIZE, 0)

index 8fc49ef..8947014 100644
@@ -1557,8 +1557,6 @@ class FileLayoutFile(object): # XXX This should inherit from fs_base.py
         vol = FilelayoutVolWrapper(self._obj, device.list[index])
         return vol, v_pos, remaining
 
-import nfs4_ops as op
-
 class FilelayoutVolWrapper(object):
     def __init__(self, obj, dataserver):
         self._obj = obj

61 lines not on this page · raw message

index 4ae884a..6912565 100644
@@ -5,7 +5,7 @@ from nfs4lib import NFS4Error, NFS4Replay, inc_u32
 from xdrdef.nfs4_type import *
 from xdrdef.nfs4_const import *
 from xdrdef.sctrl_pack import SCTRLPacker, SCTRLUnpacker
-import nfs4_ops as op
+import nfs_ops
 import time, struct
 import threading
 import hmac

@@ -20,6 +20,8 @@ logging.basicConfig(level=logging.INFO,
                     format="%(levelname)-7s:%(name)s:%(message)s")
 log_cb = logging.getLogger("nfs.client.cb")
 
+op4 = nfs_ops.NFS4ops()
+
 class NFS4Client(rpc.Client, rpc.Server):
     def __init__(self, host='localhost', port=2049, minorversion=1, ctrl_proc=16, summary=None):
         rpc.Client.__init__(self, 100003, 4)

@@ -275,7 +277,7 @@ class NFS4Client(rpc.Client, rpc.Server):
         owner = client_owner4(verf, name)
         if protect is None:
             protect = state_protect4_a(SP4_NONE)
-        res = self.compound([op.exchange_id(owner, flags, protect,
+        res = self.compound([op4.exchange_id(owner, flags, protect,
                                             [self.impl_id])],
                             cred)
         nfs4lib.check(res, expect)

@@ -287,7 +289,7 @@ class NFS4Client(rpc.Client, rpc.Server):
     def new_client_session(self, name, flags=0, sec=None):
         c = self.new_client(name, flags=flags)
         s = c.create_session(sec=sec)
-        s.compound([op.reclaim_complete(FALSE)])
+        s.compound([op4.reclaim_complete(FALSE)])
         return s
 
 class ClientStateProtection(object):

@@ -339,7 +341,7 @@ class ClientRecord(object):
         if prog is None:
             prog = self.c.prog
         for item in xrange(max_retries):
-            res = self.c.compound([op.create_session(self.clientid, self.seqid,
+            res = self.c.compound([op4.create_session(self.clientid, self.seqid,
                                                  flags,
                                                  fore_attrs, back_attrs,
                                                  prog, sec)],

@@ -429,7 +431,7 @@ class SessionRecord(object):
             raise RuntimeError
             slot = self.fore_channel.slots[slot]
         # STUB, need to properly set highest
-        return op.sequence(self.sessionid, slot.get_seqid(seq_delta),
+        return op4.sequence(self.sessionid, slot.get_seqid(seq_delta),
                            slot.id, slot.id, cache_this)
 
     def set_ssv(self, ssv=None, *args, **kwargs):

@@ -442,7 +444,7 @@ class SessionRecord(object):
         p = nfs4lib.FancyNFS4Packer()
         p.pack_SEQUENCE4args(seq_op.opsequence)
         digest =  protect.context.hmac(p.get_buffer(), SSV4_SUBKEY_MIC_I2T)
-        ssv_op = op.set_ssv(ssv, digest)
+        ssv_op = op4.set_ssv(ssv, digest)
         res = self.c.compound([seq_op, ssv_op], *args, **kwargs)
         # STUB - do some checking
         protect.context.set_ssv(ssv)

index 116324a..02352e1 100644
@@ -3,7 +3,7 @@ import rpc
 import xdrdef.nfs4_const
 from xdrdef.nfs4_pack import NFS4Packer, NFS4Unpacker
 import xdrdef.nfs4_type
-import nfs4_ops as op
+import nfs_ops
 import time
 import collections
 import hmac

@@ -30,6 +30,8 @@ state01 = xdrdef.nfs4_type.stateid4(1, "\0" * 12)
 
 import hashlib # Note this requires 2.5 or higher
 
+op4 = nfs_ops.NFS4ops()
+
 # Note that all the oid strings have tag and length bytes prepended, as
 # per description of sec_oid4 in draft26 sect 3.2
 

@@ -626,9 +628,9 @@ def use_obj(file):
     if file is None or file == [None]:
         return []
     elif type(file) is str:
-        return [op.putfh(file)]
+        return [op4.putfh(file)]
     else:
-        return [op.putrootfh()] + [op.lookup(comp) for comp in file]
+        return [op4.putrootfh()] + [op4.lookup(comp) for comp in file]
 
 ###############################################
 # Attribute information

index 2f3cd59..2214c0d 100644
@@ -8,12 +8,14 @@ from nfs4lib import NFS4Error
 #from xdrdef.nfs4_type import stateid4
 from xdrdef.nfs4_type import *
 from xdrdef.nfs4_const import *
-import nfs4_ops as op
+import nfs_ops
 import rpc
 import logging
 
 log = logging.getLogger("nfs.server.state")
 
+op4 = nfs_ops.NFS4ops()
+
 POSIXLOCK = False
 
 SHARE, BYTE, DELEG, LAYOUT, ANON = range(5) # State types

@@ -748,9 +750,9 @@ class DelegEntry(StateTableEntry):
         # ANSWER - we care about self.status, which can be set to 
         # INVALID anytime by deleg_return
         slot = session.channel_back.choose_slot()
-        seq_op = op.cb_sequence(session.sessionid, slot.get_seqid(),
+        seq_op = op4.cb_sequence(session.sessionid, slot.get_seqid(),
                                 slot.id, slot.id, True, []) # STUB
-        recall_op = op.cb_recall(self.get_id(cb=True), False, self.file.fh)
+        recall_op = op4.cb_recall(self.get_id(cb=True), False, self.file.fh)
         if self.invalid:
             # Race here doesn't matter, but would like to avoid the
             # RPC if possible.

89 lines not on this page · raw message

new file mode 100644
index 0000000..b632322
@@ -0,0 +1,68 @@
+const MNTPATHLEN = 1024;   /* Maximum bytes in a path name */
+const MNTNAMLEN = 255;     /* Maximum bytes in a name */
+const FHSIZE3 = 64;        /* Maximum bytes in a V3 file handle */
+
+typedef opaque  fhandle3 <FHSIZE3>;
+typedef string  dirpath <MNTPATHLEN>;
+typedef string  name <MNTNAMLEN>;
+
+struct mountres3_ok {
+        fhandle3        fhandle;
+        int             auth_flavors <>;
+};
+
+union mountres3 switch (mountstat3 fhs_status) {
+case MNT3_OK:
+        mountres3_ok mountinfo;
+default:
+        void;
+};
+
+struct mountbody {
+        name            ml_hostname;
+        dirpath         ml_directory;
+        mountlist       ml_next;
+};
+
+typedef mountbody *mountlist;
+
+
+struct groupnode {
+        name            gr_name;
+        groups          gr_next;
+};
+
+typedef groupnode *groups;
+
+
+struct exportnode {
+        dirpath         ex_dir;
+        groups          ex_groups;
+        exports         ex_next;
+};
+
+typedef exportnode *exports;
+
+enum mountstat3 {
+        MNT3_OK = 0,                     /* no error */
+        MNT3ERR_PERM = 1,                /* Not owner */
+        MNT3ERR_NOENT = 2,               /* No such file or directory */
+        MNT3ERR_IO = 5,                  /* I/O error */
+        MNT3ERR_ACCES = 13,              /* Permission denied */
+        MNT3ERR_NOTDIR = 20,             /* Not a directory */
+        MNT3ERR_INVAL = 22,              /* Invalid argument */
+        MNT3ERR_NAMETOOLONG = 63,        /* Filename too long */
+        MNT3ERR_NOTSUPP = 10004,         /* Operation not supported */
+        MNT3ERR_SERVERFAULT = 10006      /* A failure on the server */
+};
+
+program MOUNT_PROGRAM {
+        version MOUNT_V3 {
+                void            MOUNTPROC3_NULL(void) = 0;
+                mountres3       MOUNTPROC3_MNT(dirpath) = 1;
+                mountlist       MOUNTPROC3_DUMP(void) = 2;
+                void            MOUNTPROC3_UMNT(dirpath) = 3;
+                void            MOUNTPROC3_UMNTALL(void) = 4;
+                exports         MOUNTPROC3_EXPORT(void) = 5;
+        } = 3;
+} = 100005;

818 lines not on this page · raw message

51 lines not on this page · raw message

index d092cf7..5c604dc 100644
@@ -15,7 +15,7 @@ log = logging.getLogger("Dataserver Manager")
 
 op4 = nfs_ops.NFS4ops()
 
-class DataServer41(object):
+class DataServer(object):
     def __init__(self, server, port, path, flavor=rpc.AUTH_SYS, active=True, mdsds=True, multipath_servers=None, summary=None):
         self.mdsds = mdsds
         self.server = server

@@ -43,29 +43,47 @@ class DataServer41(object):
         self.active = True
         if not self.mdsds:
             self.connect()
+            self.make_root()
 
     def down(self):
+        self.disconnect()
         self.active = False
 
+    def reset(self):
+        self.down()
+        self.up()
+
+    def get_netaddr4(self):
+        # STUB server multipathing not supported yet
+        uaddr = '.'.join([self.server,
+                          str(self.port >> 8),
+                          str(self.port & 0xff)])
+        return type4.netaddr4(self.proto, uaddr)
+
+    def get_multipath_netaddr4s(self):
+        netaddr4s = []
+        for addr in self.multipath_servers:
+            server, port = addr
+            uaddr = '.'.join([server,
+                            str(port >> 8),
+                            str(port & 0xff)])
+            proto = "tcp"
+            if server.find(':') >= 0:
+                proto = "tcp6"
+
+            netaddr4s.append(type4.netaddr4(proto, uaddr))
+        return netaddr4s
+
+    def fh_to_name(self, mds_fh):
+        return hashlib.sha1("%r" % mds_fh).hexdigest()
+
     def connect(self):
-        # only support root with AUTH_SYS for now
-        s1 = rpc.security.instance(rpc.AUTH_SYS)
-        self.cred1 = s1.init_cred(uid=0, gid=0)
-        self.c1 = nfs4client.NFS4Client(self.server, self.port,
-                                        summary=self.summary)
-        self.c1.set_cred(self.cred1)
-        self.c1.null()
-        c = self.c1.new_client("DS.init_%s" % self.server)
-        # This is a hack to ensure MDS/DS communication path is at least
-        # as wide as the client/MDS channel (at least for linux client)
-        fore_attrs = type4.channel_attrs4(0, 16384, 16384, 2868, 8, 8, [])
-        self.sess = c.create_session(fore_attrs=fore_attrs)
-        sess.compound([op.reclaim_complete(FALSE)])
-        self.make_root()
+        raise NotImplemented
 
     def disconnect(self):
         pass
 
+class DataServer41(DataServer):
     def _execute(self, ops, exceptions=[], delay=5, maxretries=3):
         """ execute the NFS call
         If an error code is specified in the exceptions it means that the

@@ -86,36 +104,29 @@ class DataServer41(object):
                     log.error("Too many retries with DS %s" % self.server)
                     raise Exception("Dataserver communication retry error")
             elif res.status in state_errors:
-                self.disconnect()
-                self.connect()
+                self.reset()
             else:
                 log.error("Unhandled status %s from DS %s" %
                           (nfsstat4[res.status], self.server))
                 raise Exception("Dataserver communication error")
 
-    def get_netaddr4(self):
-        # STUB server multipathing not supported yet
-        uaddr = '.'.join([self.server,
-                          str(self.port >> 8),
-                          str(self.port & 0xff)])
-        return type4.netaddr4(self.proto, uaddr)
-
-    def get_multipath_netaddr4s(self):
-        netaddr4s = []
-        for addr in self.multipath_servers:
-            server, port = addr
-            uaddr = '.'.join([server,
-                            str(port >> 8),
-                            str(port & 0xff)])
-            proto = "tcp"
-            if server.find(':') >= 0:
-                proto = "tcp6"
-
-            netaddr4s.append(type4.netaddr4(proto, uaddr))
-        return netaddr4s
-
+    def connect(self):
+        # only support root with AUTH_SYS for now
+        s1 = rpc.security.instance(rpc.AUTH_SYS)
+        self.cred1 = s1.init_cred(uid=0, gid=0)
+        self.c1 = nfs4client.NFS4Client(self.server, self.port,
+                                        summary=self.summary)
+        self.c1.set_cred(self.cred1)
+        self.c1.null()
+        c = self.c1.new_client("DS.init_%s" % self.server)
+        # This is a hack to ensure MDS/DS communication path is at least
+        # as wide as the client/MDS channel (at least for linux client)
+        fore_attrs = type4.channel_attrs4(0, 16384, 16384, 2868, 8, 8, [])
+        self.sess = c.create_session(fore_attrs=fore_attrs)
+        self.sess.compound([op4.reclaim_complete(const4.FALSE)])
 
-    def make_root(self, attrs={const4.FATTR4_MODE:0777}):
+    def make_root(self):
+        attrs = {const4.FATTR4_MODE:0777}
         existing_path = []
         kind = type4.createtype4(const4.NF4DIR)
         for comp in self.path:

@@ -134,12 +145,13 @@ class DataServer41(object):
             raise RuntimeError
         # XXX clean DS directory
 
-    def fh_to_name(self, mds_fh):
-        return hashlib.sha1("%r" % mds_fh).hexdigest()
-
-    def open_file(self, mds_fh, seqid=0,
-                  access=const4.OPEN4_SHARE_ACCESS_BOTH, deny=const4.OPEN4_SHARE_DENY_NONE,
-                  attrs={const4.FATTR4_MODE: 0777}, owner = "mds", mode=const4.GUARDED4):
+    def open_file(self, mds_fh):
+        seqid=0
+        access = const4.OPEN4_SHARE_ACCESS_BOTH
+        deny = const4.OPEN4_SHARE_DENY_NONE
+        attrs = {const4.FATTR4_MODE: 0777}
+        owner = "mds"
+        mode = const4.GUARDED4
         verifier = self.sess.c.verifier
         openflag = type4.openflag4(const4.OPEN4_CREATE, type4.createhow4(mode, attrs, verifier))
         name = self.fh_to_name(mds_fh)

index 5c604dc..f32f3d7 100644
@@ -3,9 +3,12 @@ import nfs4lib
 import xdrdef.nfs4_type as type4
 from xdrdef.nfs4_pack import NFS4Packer
 import xdrdef.nfs4_const as const4
+import xdrdef.nfs3_type as type3
+import xdrdef.nfs3_const as const3
 import time
 import logging
 import nfs4client
+import nfs3client
 import hashlib
 import sys
 import nfs_ops

@@ -14,6 +17,7 @@ import socket
 log = logging.getLogger("Dataserver Manager")
 
 op4 = nfs_ops.NFS4ops()
+op3 = nfs_ops.NFS3ops()
 
 class DataServer(object):
     def __init__(self, server, port, path, flavor=rpc.AUTH_SYS, active=True, mdsds=True, multipath_servers=None, summary=None):

@@ -207,6 +211,102 @@ class DataServer41(DataServer):
         attrdict = res.resarray[-1].obj_attributes
         return attrdict.get(const4.FATTR4_SIZE, 0)
 
+class DataServer3(DataServer):
+    def _execute(self, procnum, procarg, exceptions=(), delay=5, maxretries=3):
+        """ execute the NFS call
+        If an error code is specified in the exceptions it means that the
+        caller wants to handle the error himself
+        """
+        retry_errors = []
+        while True:
+            res = self.c1.proc(procnum, procarg)
+            if res.status == const3.NFS3_OK or res.status in exceptions:
+                return res
+            elif res.status in retry_errors:
+                if maxretries > 0:
+                    maxretries -= 1
+                    time.sleep(delay)
+                else:
+                    log.error("Too many retries with DS %s" % self.server)
+                    raise Exception("Dataserver communication retry error")
+            else:
+                log.error("Unhandled status %s from DS %s" %
+                          (const3.nfsstat3[res.status], self.server))
+                raise Exception("Dataserver communication error")
+
+    def connect(self):
+        # only support root with AUTH_SYS for now
+        s1 = rpc.security.instance(rpc.AUTH_SYS)
+        self.cred1 = s1.init_cred(uid=0, gid=0)
+        self.c1 = nfs3client.NFS3Client(self.server, self.port,
+                                        summary=self.summary)
+        self.c1.set_cred(self.cred1)
+        self.rootfh = type3.nfs_fh3(self.c1.mntclnt.get_rootfh(self.path))
+        self.c1.null()
+
+    def make_root(self):
+        """ don't actually make a root path - we must use it as the export """
+        need = const3.ACCESS3_READ | const3.ACCESS3_LOOKUP | \
+               const3.ACCESS3_MODIFY | const3.ACCESS3_EXTEND
+        arg = op3.access(self.rootfh, need)
+        res = self._execute(const3.NFSPROC3_ACCESS, arg)
+        if res.resok.access != need:
+            raise RuntimeError
+        # XXX clean DS directory
+
+    def open_file(self, mds_fh):
+        name = self.fh_to_name(mds_fh)
+        where = type3.diropargs3(self.rootfh, name)
+        attr = type3.sattr3(mode=type3.set_mode3(True, 0777),
+                            uid=type3.set_uid3(True, 0),
+                            gid=type3.set_gid3(True, 0),
+                            size=type3.set_size3(False),
+                            atime=type3.set_atime(False),
+                            mtime=type3.set_mtime(False))
+        how = type3.createhow3(const3.GUARDED, attr)
+        arg = op3.create(where, how)
+        res = self._execute(const3.NFSPROC3_CREATE, arg,
+                            exceptions=(const3.NFS3ERR_EXIST,))
+
+        if res.status == const3.NFS3_OK:
+            self.filehandles[mds_fh] = (res.resok.obj.handle, None)
+
+        else:
+            arg = op3.lookup(type3.diropargs3(self.rootfh, name))
+            res = self._execute(const3.NFSPROC3_LOOKUP, arg)
+
+            self.filehandles[mds_fh] = (res.resok.object, None)
+
+    def close_file(self, mds_fh):
+        del self.filehandles[mds_fh]
+
+    def read(self, fh, pos, count):
+        arg = op3.read(fh, pos, count)
+        res = self._execute(const3.NFSPROC3_READ, arg)
+        # XXX check res.status?
+        return res.resok.data
+
+    def write(self, fh, pos, data):
+        arg = op3.write(fh, pos, len(data), const3.FILE_SYNC, data)
+        # There are all sorts of error handling issues here
+        res = self._execute(const3.NFSPROC3_WRITE, arg)
+
+    def truncate(self, fh, size):
+        attr = type3.sattr3(mode=type3.set_mode3(False),
+                            uid=type3.set_uid3(False),
+                            gid=type3.set_gid3(False),
+                            size=type3.set_size3(True, size),
+                            atime=type3.set_atime(False),
+                            mtime=type3.set_mtime(False))
+        arg = op3.setattr(fh, attr, type3.sattrguard3(check=False))
+        res = self._execute(const3.NFSPROC3_SETATTR, arg)
+
+    def get_size(self, fh):
+        arg = op3.getattr(fh)
+        res = self._execute(const3.NFSPROC3_GETATTR, arg)
+        # XXX check res.status?
+        return res.resok.obj_attributes.size
+
 
 class DSDevice(object):
     def __init__(self, mdsds):

180 lines not on this page · raw message

index 3d2d4cf..7ca4d92 100755
@@ -1867,27 +1867,6 @@ class NFS4Server(rpc.Server):
                                   new_cookie >= len(list))
         return encode_status(NFS4_OK, res)
 
-#     def op_getdeviceinfo(self, arg, env): # STUB
-#         check_session(env)
-#         # check_cfh(env)
-#         # fs = env.cfh.fs
-#         # STUB - only deals with block volumes
-#         kind = arg.gdia_layout_type
-#         if kind != LAYOUT4_BLOCK_VOLUME:
-#             return encode_status(NFS4ERR_INVAL)
-#         # STUB - want to pull this from fs, not block module
-#         d = block.devices.get(arg.gdia_device_id, None)
-#         if d is None:
-#             return encode_status(NFS4ERR_INVAL)
-#         address = device_addr4(LAYOUT4_BLOCK_VOLUME, d.get_addr())
-#         # Check that we don't exceed count
-#         p = nfs4lib.FancyNFS4Packer()
-#         p.pack_device_addr4(address)
-#         if len(p.get_buffer()) > arg.gdia_maxcount:
-#             return encode_status(NFS4ERR_TOOSMALL, gdir_mincount = len(p.get_buffer()))
-#         res = GETDEVICEINFO4resok(address, 0)
-#         return encode_status(NFS4_OK, res)
-
     def op_getdeviceinfo(self, arg, env): # STUB
         # STUB - ignoring notifications
         check_session(env)

index 250e945..4801883 100644
@@ -351,6 +351,7 @@ class RpcPipe(Pipe):
         self._pending = {} # {xid:defer}
         self._lock = threading.Lock() # Protects fields below
         self._xid = 0
+        self.set_active()
 
     def _get_xid(self):
         with self._lock:

@@ -358,6 +359,15 @@ class RpcPipe(Pipe):
             self._xid = inc_u32(out)
         return out
 
+    def set_active(self):
+        self._active = True
+
+    def clear_active(self):
+        self._active = False
+
+    def is_active(self):
+        return self._active
+
     def listen(self, xid, timeout=None):
         """Wait for a reply to a CALL."""
         self._pending[xid].wait(timeout)

@@ -500,15 +510,24 @@ class ConnectionHandler(object):
                 log_p.warn(1, "polling error from %i" % fd)
                 # STUB - now what?
             for fd in w:
-                self._event_write(fd)
+                try:
+                    self._event_write(fd)
+                except socket.error, e:
+                    self._event_close(fd)
             for fd in r:
                 if fd in self.listeners:
-                    self._event_connect_incoming(fd)
+                    try:
+                        self._event_connect_incoming(fd)
+                    except socket.error, e:
+                        self._event_close(fd)
                 elif fd == self._alarm_poll.fileno():
                     commands = self._alarm_poll.recv(self.rsize)
                     for c in commands:
                         data = self._alarm.pop()
-                        switch[c](data)
+                        try:
+                            switch[c](data)
+                        except socket.error, e:
+                            self._event_close(fd)
                 else:
                     try:
                         data = self.sockets[fd].recv_records(self.rsize)

@@ -557,6 +576,7 @@ class ConnectionHandler(object):
         self.writelist -= temp
         self.readlist -= temp
         self.errlist -= temp
+        self.sockets[fd].clear_active()
         self.sockets[fd].close()
         del self.sockets[fd]
 

index 79a6f0e..176765c 100644
@@ -30,14 +30,19 @@ class PORTMAPClient(rpc.Client):
     def __init__(self, host='localhost', port=PMAP_PORT):
         rpc.Client.__init__(self, PMAP_PROG, PMAP_VERS)
         self.server_address = (host, port)
-        self.c1 = self.connect(self.server_address)
+        self._pipe = None
+
+    def get_pipe(self):
+        if not self._pipe or not self._pipe.is_active():
+           self._pipe = self.connect(self.server_address)
+        return self._pipe
 
     def proc_async(self, procnum, procarg, credinfo=None, pipe=None,
                    checks=True, packer=PORTMAPPacker):
         if credinfo is None:
             credinfo = self.default_cred
         if pipe is None:
-            pipe = self.c1
+            pipe = self.get_pipe()
         p = packer(check_enum=checks, check_array=checks)
         arg_packer = getattr(p, 'pack_%s' % procarg.__class__.__name__)
         arg_packer(procarg)

@@ -51,7 +56,7 @@ class PORTMAPClient(rpc.Client):
 
     def listen(self, xid, restypename, pipe=None, timeout=10.0):
         if pipe is None:
-            pipe = self.c1
+            pipe = self.get_pipe()
         header, data = pipe.listen(xid, timeout)
         if data:
             p = PORTMAPUnpacker(data)

@@ -69,14 +74,19 @@ class Mnt3Client(rpc.Client):
     def __init__(self, host='localhost', port=None):
         rpc.Client.__init__(self, MOUNT_PROGRAM, MOUNT_V3)
         self.server_address = (host, port)
-        self.c1 = self.connect(self.server_address)
+        self._pipe = None
+
+    def get_pipe(self):
+        if not self._pipe or not self._pipe.is_active():
+            self._pipe = self.connect(self.server_address)
+        return self._pipe
 
     def proc_async(self, procnum, procarg, credinfo=None, pipe=None,
                    checks=True, packer=MNT3Packer):
         if credinfo is None:
             credinfo = self.default_cred
         if pipe is None:
-            pipe = self.c1
+            pipe = self.get_pipe()
         p = packer(check_enum=checks, check_array=checks)
         arg_packer = getattr(p, 'pack_%s' % procarg.__class__.__name__)
         arg_packer(procarg)

@@ -90,7 +100,7 @@ class Mnt3Client(rpc.Client):
 
     def listen(self, xid, restypename, pipe=None, timeout=10.0):
         if pipe is None:
-            pipe = self.c1
+            pipe = self.get_pipe()
         header, data = pipe.listen(xid, timeout)
         if data:
             p = MNT3Unpacker(data)

@@ -110,15 +120,6 @@ class Mnt3Client(rpc.Client):
 class NFS3Client(rpc.Client):
     def __init__(self, host='localhost', port=None, ctrl_proc=16, summary=None):
         rpc.Client.__init__(self, 100003, 3)
-        #self.prog = 0x40000000
-        #self.versions = [1] # List of supported versions of prog
-
-        #self.minorversion = minorversion
-        #self.minor_versions = [minorversion]
-        #self.tag = "default tag"
-        #self.impl_id = nfs_impl_id4("citi.umich.edu", "pynfs X.X",
-        #                            nfs4lib.get_nfstime())
-
         self.portmap = PORTMAPClient(host=host)
         self.mntport = self.portmap.get_port(MOUNT_PROGRAM, MOUNT_V3)
         if not port:

@@ -128,17 +129,21 @@ class NFS3Client(rpc.Client):
 
         self.verifier = struct.pack('>d', time.time())
         self.server_address = (host, self.port)
-        self.c1 = self.connect(self.server_address)
-        #self.sessions = {} # XXX Really, this should be per server
         self.ctrl_proc = ctrl_proc
         self.summary = summary
+        self._pipe = None
         self.mntclnt = Mnt3Client(host=host, port=self.mntport)
 
+    def get_pipe(self):
+        if not self._pipe or not self._pipe.is_active():
+            self._pipe = self.connect(self.server_address)
+        return self._pipe
+
     def set_cred(self, credinfo):
         self.default_cred = credinfo
 
     def null_async(self, data=""):
-        return self.send_call(self.c1, 0, data)
+        return self.send_call(self.get_pipe(), 0, data)
 
     def null(self, *args, **kwargs):
         xid = self.null_async(*args, **kwargs)

@@ -149,7 +154,7 @@ class NFS3Client(rpc.Client):
         if credinfo is None:
             credinfo = self.default_cred
         if pipe is None:
-            pipe = self.c1
+            pipe = self.get_pipe()
         p = packer(check_enum=checks, check_array=checks)
         arg_packer = getattr(p, 'pack_%s' % procarg.__class__.__name__)
         arg_packer(procarg)

@@ -167,7 +172,7 @@ class NFS3Client(rpc.Client):
 
     def listen(self, xid, procarg=None, pipe=None, timeout=10.0):
         if pipe is None:
-            pipe = self.c1
+            pipe = self.get_pipe()
         header, data = pipe.listen(xid, timeout)
         if data:
             p = NFS3Unpacker(data)