Sync with NEO/py v1.10

BUGS.rst

@@ -16,6 +16,19 @@ This happens in the following conditions:
 4. the cell is checked completely before it could replicate up to the max tid
    to check
 
+Sometimes, it causes the master to crash::
+
+    File "neo/lib/handler.py", line 72, in dispatch
+      method(conn, *args, **kw)
+    File "neo/master/handlers/storage.py", line 93, in notifyReplicationDone
+      cell_list = app.backup_app.notifyReplicationDone(node, offset, tid)
+    File "neo/master/backup_app.py", line 337, in notifyReplicationDone
+      assert cell.isReadable()
+  AssertionError
+
 Workaround: make sure all cells are up-to-date before checking replicas.
 
-Found by running testBackupNodeLost many times.
+Found by running testBackupNodeLost many times:
+
+- either a failureException: 12 != 11
+- or the above assert failure, in which case the unit test freezes

CHANGELOG.rst

 Change History
 ==============
 
+1.10 (2018-07-16)
+-----------------
+
+A important performance improvement is that the replication now remembers where
+it was interrupted: a storage node that gets disconnected for a short time now
+gets fully operational quite instantaneously because it only has to replicate
+the new data. Before, the time to recover depended on the size of the DB, just
+to verify that most of the data are already transferred.
+
+As a small optimization, an empty transaction extension is now serialized with
+an empty string.
+
+The above 2 changes required a bump of the protocol version, as well as an
+upgrade of the storage format. Once upgraded (this is done automatically as
+usual), databases can't be opened anymore by older versions of NEO.
+
+Other general changes:
+
+- Add support for custom compression levels.
+- Maximize resiliency by taking into account the topology of storage nodes.
+- Fix a few issues with ZODB5. Note however that merging several DB with the
+  Importer backend only works if they were only used with ZODB < 5.
+
+Master:
+
+- Automatically discard feeding cells that get out-of-date.
+
+Client:
+
+- Fix partial import from a source storage.
+- Store uncompressed if compressed size is equal.
+
+Storage:
+
+- Fixed v1.9 code that sped up the replication by sending bigger network
+  packets.
+- Fix replication of creation undone.
+- Stop logging 'Abort TXN' for txn that have been locked.
+- Clarify log about data deletion of discarded cells.
+
+MySQL backend:
+
+- Fix replication of big OIDs (> 16M).
+- Do not full-scan for duplicates of big OIDs if deduplication is disabled.
+- Fix remaining places where a server disconnection was not catched.
+
+SQlite backend:
+
+- Fix indexes of upgraded databases.
+
+Importer backend:
+
+- Fetch and process the data to import in a separate process. It is even
+  usually free to use the best compression level.
+- New option to write back new transactions to the source database.
+  See 'importer.conf' for more information.
+- Give a title to the 'import' and 'writeback' subprocesses,
+  if the 'setproctitle' egg is installed.
+- Log when the transaction index for FileStorage DB is built.
+- Open imported database in read-only whenever possible.
+- Do not trigger speedupFileStorageTxnLookup uselessly.
+- Do not checksum data twice.
+- Fix NameError when recovering during tpc_finish.
+
 1.9 (2018-03-13)
 ----------------
 

MANIFEST.in

 graft tools
-include neo.conf CHANGELOG.rst TODO TESTS.txt ZODB3.patch
+include neo.conf CHANGELOG.rst TODO ZODB3.patch

go/neo/neonet/connection_test.go

-// Copyright (C) 2016-2018  Nexedi SA and Contributors.
+// Copyright (C) 2016-2020  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -1015,8 +1015,8 @@ func benchmarkLinkRTT(b *testing.B, l1, l2 *NodeLink) {
 			case *proto.GetObject:
 				err = req.Reply(&proto.AnswerObject{
 					Oid:        msg.Oid,
-					Serial:     msg.Serial,
-					DataSerial: msg.Tid,
+					Serial:     msg.At,
+					DataSerial: msg.Before,
 				})
 				if err != nil {
 					b.Fatal(err)
@@ -1033,15 +1033,15 @@ func benchmarkLinkRTT(b *testing.B, l1, l2 *NodeLink) {
 		obj := &proto.AnswerObject{}
 
 		get.Oid = zodb.Oid(i)
-		get.Serial = zodb.Tid(i+1)
-		get.Tid = zodb.Tid(i+2)
+		get.At = zodb.Tid(i+1)
+		get.Before = zodb.Tid(i+2)
 
 		err := l1.Ask1(get, obj)
 		if err != nil {
 			b.Fatal(err)
 		}
 
-		if !(obj.Oid == get.Oid && obj.Serial == get.Serial && obj.DataSerial == get.Tid) {
+		if !(obj.Oid == get.Oid && obj.Serial == get.At && obj.DataSerial == get.Before) {
 			b.Fatalf("read back: %v  ; requested %v", obj, get)
 		}
 

go/neo/proto/proto_test.go

-// Copyright (C) 2016-2017  Nexedi SA and Contributors.
+// Copyright (C) 2016-2020  Nexedi SA and Contributors.
 //                          Kirill Smelkov <kirr@nexedi.com>
 //
 // This program is free software: you can Use, Study, Modify and Redistribute
@@ -42,6 +42,11 @@ func hex(s string) string {
 	return string(b)
 }
 
+// uint8 -> string as encoded on the wire
+func u8(v uint8) string {
+	return string(v)
+}
+
 // uint16 -> string as encoded on the wire
 func u16(v uint16) string {
 	var b [2]byte
@@ -163,8 +168,8 @@ func TestMsgMarshal(t *testing.T) {
 		// empty
 		{&Ping{}, ""},
 
-		// uint32, string
-		{&Error{Code: 0x01020304, Message: "hello"}, "\x01\x02\x03\x04\x00\x00\x00\x05hello"},
+		// uint8, string
+		{&Error{Code: 0x04, Message: "hello"}, "\x04\x00\x00\x00\x05hello"},
 
 		// Oid, Tid, bool, Checksum, []byte
 		{&StoreObject{
@@ -194,9 +199,9 @@ func TestMsgMarshal(t *testing.T) {
 
 			hex("0102030405060708") +
 			hex("00000003") +
-				hex("00000001000000020000000b000000000000001100000001") +
-				hex("00000002000000010000000b00000002") +
-				hex("00000007000000030000000b000000040000000f000000030000001700000000"),
+				hex("00000001000000020000000b010000001100") +
+				hex("00000002000000010000000b02") +
+				hex("00000007000000030000000b030000000f040000001701"),
 		},
 
 		// map[Oid]struct {Tid,Tid,bool}
@@ -245,11 +250,12 @@ func TestMsgMarshal(t *testing.T) {
 		},
 
 		// uint32, Address, string, IdTime
-		{&RequestIdentification{CLIENT, 17, Address{"localhost", 7777}, "myname", 0.12345678},
+		{&RequestIdentification{CLIENT, 17, Address{"localhost", 7777}, "myname", []string{"room1", "rack234"}, 0.12345678},
 
-			u32(2) + u32(17) + u32(9) +
+			u8(2) + u32(17) + u32(9) +
 			"localhost" + u16(7777) +
 			u32(6) + "myname" +
+			u32(2) + u32(5)+"room1" + u32(7)+"rack234" +
 			hex("3fbf9add1091c895"),
 		},
 
@@ -258,7 +264,7 @@ func TestMsgMarshal(t *testing.T) {
 			{CLIENT, Address{}, UUID(CLIENT, 1), RUNNING, 1504466245.925599}}},
 
 			hex("41d66b15517b469d") + u32(1) +
-				u32(2) + u32(0) /* <- ø Address */ + hex("e0000001") + u32(2) +
+				u8(2) + u32(0) /* <- ø Address */ + hex("e0000001") + u8(2) +
 				hex("41d66b15517b3d04"),
 		},
 

go/neo/proto/protogen.go

@@ -341,14 +341,12 @@ import (
 			// generate code for this type to implement neo.Msg
 			var msgCode MsgCode
 			msgCode.answer = specAnnotation.answer || strings.HasPrefix(typename, "Answer")
-			switch {
-			case !msgCode.answer || typename == "Error":
-				msgCode.msgSerial = msgSerial
-
-			// answer to something
-			default:
-				msgCode.msgSerial = msgSerial - 1
+			// increment msgSerial only by +1 when going from
+			// request1->request2 in `Request1 Answer1 Request2`.
+			if msgCode.answer && typename != "Error" {
+				msgSerial--
 			}
+			msgCode.msgSerial = msgSerial
 
 			fmt.Fprintf(&buf, "// %s. %s\n\n", msgCode, typename)
 

go/neo/proto/zproto-str.go

@@ -27,7 +27,7 @@ const _ErrorCode_name = "ACKNOT_READYOID_NOT_FOUNDTID_NOT_FOUNDOID_DOES_NOT_EXIS
 var _ErrorCode_index = [...]uint8{0, 3, 12, 25, 38, 56, 70, 87, 101, 124, 141, 157, 179}
 
 func (i ErrorCode) String() string {
-	if i >= ErrorCode(len(_ErrorCode_index)-1) {
+	if i < 0 || i >= ErrorCode(len(_ErrorCode_index)-1) {
 		return "ErrorCode(" + strconv.FormatInt(int64(i), 10) + ")"
 	}
 	return _ErrorCode_name[_ErrorCode_index[i]:_ErrorCode_index[i+1]]
@@ -99,16 +99,16 @@ func _() {
 	// An "invalid array index" compiler error signifies that the constant values have changed.
 	// Re-run the stringer command to generate them again.
 	var x [1]struct{}
-	_ = x[UP_TO_DATE-0]
-	_ = x[OUT_OF_DATE-1]
+	_ = x[OUT_OF_DATE-0]
+	_ = x[UP_TO_DATE-1]
 	_ = x[FEEDING-2]
-	_ = x[DISCARDED-3]
-	_ = x[CORRUPTED-4]
+	_ = x[CORRUPTED-3]
+	_ = x[DISCARDED-4]
 }
 
-const _CellState_name = "UP_TO_DATEOUT_OF_DATEFEEDINGDISCARDEDCORRUPTED"
+const _CellState_name = "OUT_OF_DATEUP_TO_DATEFEEDINGCORRUPTEDDISCARDED"
 
-var _CellState_index = [...]uint8{0, 10, 21, 28, 37, 46}
+var _CellState_index = [...]uint8{0, 11, 21, 28, 37, 46}
 
 func (i CellState) String() string {
 	if i < 0 || i >= CellState(len(_CellState_index)-1) {

go/neo/proto/ztestdata_proto_py_test.go

@@ -3,102 +3,102 @@ package proto
 
 var pyMsgRegistry = map[uint16]string{
 	1:	"RequestIdentification",
-	3:	"Ping",
-	5:	"CloseClient",
-	6:	"PrimaryMaster",
-	8:	"NotPrimaryMaster",
-	9:	"NotifyNodeInformation",
-	10:	"Recovery",
-	12:	"LastIDs",
-	14:	"AskPartitionTable",
-	16:	"SendPartitionTable",
-	17:	"NotifyPartitionChanges",
-	18:	"StartOperation",
-	19:	"StopOperation",
-	20:	"UnfinishedTransactions",
-	22:	"LockedTransactions",
-	24:	"FinalTID",
-	26:	"ValidateTransaction",
-	27:	"BeginTransaction",
-	29:	"FailedVote",
-	30:	"FinishTransaction",
-	32:	"LockInformation",
-	34:	"InvalidateObjects",
-	35:	"NotifyUnlockInformation",
-	36:	"AskNewOIDs",
-	38:	"NotifyDeadlock",
-	39:	"RebaseTransaction",
-	41:	"RebaseObject",
-	43:	"StoreObject",
-	45:	"AbortTransaction",
-	46:	"StoreTransaction",
-	48:	"VoteTransaction",
-	50:	"GetObject",
-	52:	"AskTIDs",
-	54:	"TransactionInformation",
-	56:	"ObjectHistory",
-	58:	"PartitionList",
-	60:	"NodeList",
-	62:	"SetNodeState",
-	63:	"AddPendingNodes",
-	64:	"TweakPartitionTable",
-	65:	"SetClusterState",
-	66:	"Repair",
-	67:	"RepairOne",
-	68:	"NotifyClusterState",
-	69:	"AskClusterState",
-	71:	"ObjectUndoSerial",
-	73:	"AskTIDsFrom",
-	75:	"Pack",
-	77:	"CheckReplicas",
-	78:	"CheckPartition",
-	79:	"CheckTIDRange",
-	81:	"CheckSerialRange",
-	83:	"PartitionCorrupted",
-	84:	"NotifyReady",
-	85:	"LastTransaction",
-	87:	"CheckCurrentSerial",
-	89:	"NotifyTransactionFinished",
-	90:	"Replicate",
-	91:	"ReplicationDone",
-	92:	"FetchTransactions",
-	94:	"FetchObjects",
-	96:	"AddTransaction",
-	97:	"AddObject",
-	98:	"Truncate",
+	2:	"Ping",
+	3:	"CloseClient",
+	4:	"PrimaryMaster",
+	5:	"NotPrimaryMaster",
+	6:	"NotifyNodeInformation",
+	7:	"Recovery",
+	8:	"LastIDs",
+	9:	"AskPartitionTable",
+	10:	"SendPartitionTable",
+	11:	"NotifyPartitionChanges",
+	12:	"StartOperation",
+	13:	"StopOperation",
+	14:	"UnfinishedTransactions",
+	15:	"LockedTransactions",
+	16:	"FinalTID",
+	17:	"ValidateTransaction",
+	18:	"BeginTransaction",
+	19:	"FailedVote",
+	20:	"FinishTransaction",
+	21:	"LockInformation",
+	22:	"InvalidateObjects",
+	23:	"NotifyUnlockInformation",
+	24:	"AskNewOIDs",
+	25:	"NotifyDeadlock",
+	26:	"RebaseTransaction",
+	27:	"RebaseObject",
+	28:	"StoreObject",
+	29:	"AbortTransaction",
+	30:	"StoreTransaction",
+	31:	"VoteTransaction",
+	32:	"GetObject",
+	33:	"AskTIDs",
+	34:	"TransactionInformation",
+	35:	"ObjectHistory",
+	36:	"PartitionList",
+	37:	"NodeList",
+	38:	"SetNodeState",
+	39:	"AddPendingNodes",
+	40:	"TweakPartitionTable",
+	41:	"SetClusterState",
+	42:	"Repair",
+	43:	"RepairOne",
+	44:	"NotifyClusterState",
+	45:	"AskClusterState",
+	46:	"ObjectUndoSerial",
+	47:	"AskTIDsFrom",
+	48:	"Pack",
+	49:	"CheckReplicas",
+	50:	"CheckPartition",
+	51:	"CheckTIDRange",
+	52:	"CheckSerialRange",
+	53:	"PartitionCorrupted",
+	54:	"NotifyReady",
+	55:	"LastTransaction",
+	56:	"CheckCurrentSerial",
+	57:	"NotifyTransactionFinished",
+	58:	"Replicate",
+	59:	"ReplicationDone",
+	60:	"FetchTransactions",
+	61:	"FetchObjects",
+	62:	"AddTransaction",
+	63:	"AddObject",
+	64:	"Truncate",
 	32768:	"Error",
 	32769:	"AcceptIdentification",
-	32771:	"Pong",
-	32774:	"AnswerPrimary",
-	32778:	"AnswerRecovery",
-	32780:	"AnswerLastIDs",
-	32782:	"AnswerPartitionTable",
-	32788:	"AnswerUnfinishedTransactions",
-	32790:	"AnswerLockedTransactions",
-	32792:	"AnswerFinalTID",
-	32795:	"AnswerBeginTransaction",
-	32798:	"AnswerTransactionFinished",
-	32800:	"AnswerInformationLocked",
-	32804:	"AnswerNewOIDs",
-	32807:	"AnswerRebaseTransaction",
-	32809:	"AnswerRebaseObject",
-	32811:	"AnswerStoreObject",
-	32814:	"AnswerStoreTransaction",
-	32816:	"AnswerVoteTransaction",
-	32818:	"AnswerObject",
-	32820:	"AnswerTIDs",
-	32822:	"AnswerTransactionInformation",
-	32824:	"AnswerObjectHistory",
-	32826:	"AnswerPartitionList",
-	32828:	"AnswerNodeList",
-	32837:	"AnswerClusterState",
-	32839:	"AnswerObjectUndoSerial",
-	32841:	"AnswerTIDsFrom",
-	32843:	"AnswerPack",
-	32847:	"AnswerCheckTIDRange",
-	32849:	"AnswerCheckSerialRange",
-	32853:	"AnswerLastTransaction",
-	32855:	"AnswerCheckCurrentSerial",
-	32860:	"AnswerFetchTransactions",
-	32862:	"AnswerFetchObjects",
+	32770:	"Pong",
+	32772:	"AnswerPrimary",
+	32775:	"AnswerRecovery",
+	32776:	"AnswerLastIDs",
+	32777:	"AnswerPartitionTable",
+	32782:	"AnswerUnfinishedTransactions",
+	32783:	"AnswerLockedTransactions",
+	32784:	"AnswerFinalTID",
+	32786:	"AnswerBeginTransaction",
+	32788:	"AnswerTransactionFinished",
+	32789:	"AnswerInformationLocked",
+	32792:	"AnswerNewOIDs",
+	32794:	"AnswerRebaseTransaction",
+	32795:	"AnswerRebaseObject",
+	32796:	"AnswerStoreObject",
+	32798:	"AnswerStoreTransaction",
+	32799:	"AnswerVoteTransaction",
+	32800:	"AnswerObject",
+	32801:	"AnswerTIDs",
+	32802:	"AnswerTransactionInformation",
+	32803:	"AnswerObjectHistory",
+	32804:	"AnswerPartitionList",
+	32805:	"AnswerNodeList",
+	32813:	"AnswerClusterState",
+	32814:	"AnswerObjectUndoSerial",
+	32815:	"AnswerTIDsFrom",
+	32816:	"AnswerPack",
+	32819:	"AnswerCheckTIDRange",
+	32820:	"AnswerCheckSerialRange",
+	32823:	"AnswerLastTransaction",
+	32824:	"AnswerCheckCurrentSerial",
+	32828:	"AnswerFetchTransactions",
+	32829:	"AnswerFetchObjects",
 }

importer.conf

@@ -14,6 +14,8 @@
 # directly to a NEO cluster with replicas or several storage nodes.
 # Importer backend can only be used with a single storage node.
 #
+# WARNING: Merging several DB only works if they were only used with ZODB < 5.
+#
 # Here is how to proceed once this file is ready:
 # 1. Restart ZODB clients to connect to new NEO cluster (not started yet).
 # 2. Start NEO cluster (use 'neoctl -a <admin> start' command if necessary).
@@ -43,6 +45,12 @@
 # (instead of adapter=Importer & database=/path_to_this_file).
 adapter=MySQL
 database=neo
+# Keep writing back new transactions to the source database, provided it is
+# not splitted. In case of any issue, the import can be aborted without losing
+# data. Note however that it is asynchronous so don't stop the storage node
+# too quickly after the last committed transaction (e.g. check with tools like
+# fstail).
+writeback=true
 
 # The other sections are for source databases.
 [root]
@@ -50,7 +58,8 @@ database=neo
 # ZEO is possible but less efficient: ZEO servers must be stopped
 # if NEO opens FileStorage DBs directly.
 # Note that NEO uses 'new_oid' method to get the last OID, that's why the
-# source DB can't be open read-only. NEO never modifies a FileStorage DB.
+# source DB can't be open read-only. Unless 'writeback' is enabled, NEO never
+# modifies a FileStorage DB.
 storage=
   <filestorage>
     path /path/to/root.fs

neo/client/Storage.py

@@ -160,11 +160,7 @@ class Storage(BaseStorage.BaseStorage,
 
     def copyTransactionsFrom(self, source, verbose=False):
         """ Zope compliant API """
-        return self.importFrom(source)
-
-    def importFrom(self, source, start=None, stop=None, preindex=None):
-        """ Allow import only a part of the source storage """
-        return self.app.importFrom(self, source, start, stop, preindex)
+        return self.app.importFrom(self, source)
 
     def pack(self, t, referencesf, gc=False):
         if gc:

neo/client/__init__.py

@@ -44,7 +44,7 @@ def patch():
         # <patch>
         serial = self._storage.tpc_finish(transaction, callback)
         if serial is not None:
-            assert isinstance(serial, str), repr(serial)
+            assert isinstance(serial, bytes), repr(serial)
             for oid_iterator in (self._modified, self._creating):
                 for oid in oid_iterator:
                     obj = self._cache.get(oid, None)

neo/client/app.py

@@ -14,11 +14,14 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from cPickle import dumps, loads
-from zlib import compress, decompress
 import heapq
 import time
 
+try:
+    from ZODB._compat import dumps, loads, _protocol
+except ImportError:
+    from cPickle import dumps, loads
+    _protocol = 1
 from ZODB.POSException import UndoError, ConflictError, ReadConflictError
 from . import OLD_ZODB
 if OLD_ZODB:
@@ -26,6 +29,7 @@ if OLD_ZODB:
 from persistent.TimeStamp import TimeStamp
 
 from neo.lib import logging
+from neo.lib.compress import decompress_list, getCompress
 from neo.lib.protocol import NodeTypes, Packets, \
     INVALID_PARTITION, MAX_TID, ZERO_HASH, ZERO_TID
 from neo.lib.util import makeChecksum, dump
@@ -50,7 +54,6 @@ if SignalHandler:
     import signal
     SignalHandler.registerHandler(signal.SIGUSR2, logging.reopen)
 
-
 class Application(ThreadedApplication):
     """The client node application."""
 
@@ -99,7 +102,7 @@ class Application(ThreadedApplication):
         # _connecting_to_master_node is used to prevent simultaneous master
         # node connection attempts
         self._connecting_to_master_node = Lock()
-        self.compress = compress
+        self.compress = getCompress(compress)
 
     def __getattr__(self, attr):
         if attr in ('last_tid', 'pt'):
@@ -215,7 +218,7 @@ class Application(ThreadedApplication):
                         node=node,
                         dispatcher=self.dispatcher)
                 p = Packets.RequestIdentification(
-                    NodeTypes.CLIENT, self.uuid, None, self.name, None)
+                    NodeTypes.CLIENT, self.uuid, None, self.name, (), None)
                 try:
                     ask(conn, p, handler=handler)
                 except ConnectionClosed:
@@ -273,7 +276,8 @@ class Application(ThreadedApplication):
     def _askStorageForRead(self, object_id, packet, askStorage=None):
         cp = self.cp
         pt = self.pt
-        if type(object_id) is str:
+        # BBB: On Py2, it can be a subclass of bytes (binary from zodbpickle).
+        if isinstance(object_id, bytes):
             object_id = pt.getPartition(object_id)
         if askStorage is None:
             askStorage = self._askStorage
@@ -387,7 +391,7 @@ class Application(ThreadedApplication):
                     logging.error('wrong checksum from %s for oid %s',
                               conn, dump(oid))
                     raise NEOStorageReadRetry(False)
-                return (decompress(data) if compression else data,
+                return (decompress_list[compression](data),
                         tid, next_tid, data_tid)
             raise NEOStorageCreationUndoneError(dump(oid))
         return self._askStorageForRead(oid,
@@ -434,17 +438,7 @@ class Application(ThreadedApplication):
             checksum = ZERO_HASH
         else:
             assert data_serial is None
-            size = len(data)
-            if self.compress:
-                compressed_data = compress(data)
-                if size < len(compressed_data):
-                    compressed_data = data
-                    compression = 0
-                else:
-                    compression = 1
-            else:
-                compression = 0
-                compressed_data = data
+            size, compression, compressed_data = self.compress(data)
             checksum = makeChecksum(compressed_data)
             txn_context.data_size += size
         # Store object in tmp cache
@@ -553,9 +547,12 @@ class Application(ThreadedApplication):
         txn_context = self._txn_container.get(transaction)
         self.waitStoreResponses(txn_context)
         ttid = txn_context.ttid
+        ext = transaction._extension
+        ext = dumps(ext, _protocol) if ext else ''
+        # user and description are cast to str in case they're unicode.
+        # BBB: This is not required anymore with recent ZODB.
         packet = Packets.AskStoreTransaction(ttid, str(transaction.user),
-            str(transaction.description), dumps(transaction._extension),
-            txn_context.cache_dict)
+            str(transaction.description), ext, txn_context.cache_dict)
         queue = txn_context.queue
         involved_nodes = txn_context.involved_nodes
         # Ask in parallel all involved storage nodes to commit object metadata.
@@ -785,10 +782,6 @@ class Application(ThreadedApplication):
         self.waitStoreResponses(txn_context)
         return None, txn_oid_list
 
-    def _insertMetadata(self, txn_info, extension):
-        for k, v in loads(extension).items():
-            txn_info[k] = v
-
     def _getTransactionInformation(self, tid):
         return self._askStorageForRead(tid,
             Packets.AskTransactionInformation(tid))
@@ -828,7 +821,8 @@ class Application(ThreadedApplication):
             if filter is None or filter(txn_info):
                 txn_info.pop('packed')
                 txn_info.pop("oids")
-                self._insertMetadata(txn_info, txn_ext)
+                if txn_ext:
+                    txn_info.update(loads(txn_ext))
                 append(txn_info)
                 if len(undo_info) >= last - first:
                     break
@@ -856,7 +850,7 @@ class Application(ThreadedApplication):
         tid = None
         for tid in tid_list:
             (txn_info, txn_ext) = self._getTransactionInformation(tid)
-            txn_info['ext'] = loads(txn_ext)
+            txn_info['ext'] = loads(txn_ext) if txn_ext else {}
             append(txn_info)
         return (tid, txn_list)
 
@@ -875,23 +869,29 @@ class Application(ThreadedApplication):
                 txn_info['size'] = size
                 if filter is None or filter(txn_info):
                     result.append(txn_info)
-                self._insertMetadata(txn_info, txn_ext)
+                if txn_ext:
+                    txn_info.update(loads(txn_ext))
         return result
 
-    def importFrom(self, storage, source, start, stop, preindex=None):
+    def importFrom(self, storage, source):
         # TODO: The main difference with BaseStorage implementation is that
         #       preindex can't be filled with the result 'store' (tid only
         #       known after 'tpc_finish'. This method could be dropped if we
         #       implemented IStorageRestoreable (a wrapper around source would
         #       still be required for partial import).
-        if preindex is None:
-            preindex = {}
-        for transaction in source.iterator(start, stop):
+        preindex = {}
+        for transaction in source.iterator():
             tid = transaction.tid
             self.tpc_begin(storage, transaction, tid, transaction.status)
             for r in transaction:
                 oid = r.oid
-                pre = preindex.get(oid)
+                try:
+                    pre = preindex[oid]
+                except KeyError:
+                    try:
+                        pre = self.load(oid)[1]
+                    except NEOStorageNotFoundError:
+                        pre = ZERO_TID
                 self.store(oid, pre, r.data, r.version, transaction)
                 preindex[oid] = tid
             conflicted = self.tpc_vote(transaction)

neo/client/component.xml

@@ -14,10 +14,14 @@
         Give the name of the cluster
       </description>
     </key>
-    <key name="compress" datatype="boolean">
+    <key name="compress" datatype=".compress">
       <description>
-        If true, data is automatically compressed (unless compressed size is
-        not smaller). This is the default behaviour.
+        The value is either of 'boolean' type or an explicit algorithm that
+        matches the regex 'zlib(=\d+)?', where the optional number is
+        the compression level.
+        Any record that is not smaller once compressed is stored uncompressed.
+        True is the default and its meaning may change over time:
+        currently, it is the same as 'zlib'.
       </description>
     </key>
     <key name="read-only" datatype="boolean">

neo/client/config.py

@@ -23,3 +23,11 @@ class NeoStorage(BaseConfig):
         config = self.config
         return Storage(**{k: getattr(config, k)
                           for k in config.getSectionAttributes()})
+
+def compress(value):
+    from ZConfig.datatypes import asBoolean
+    try:
+        return asBoolean(value)
+    except ValueError:
+        from neo.lib.compress import parseOption
+    return parseOption(value)

neo/client/handlers/storage.py

@@ -14,10 +14,10 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-from zlib import decompress
 from ZODB.TimeStamp import TimeStamp
 
 from neo.lib import logging
+from neo.lib.compress import decompress_list
 from neo.lib.protocol import Packets, uuid_str
 from neo.lib.util import dump, makeChecksum
 from neo.lib.exception import NodeNotReady
@@ -129,8 +129,7 @@ class StorageAnswersHandler(AnswerBaseHandler):
                             'wrong checksum while getting back data for'
                             ' object %s during rebase of transaction %s'
                             % (dump(oid), dump(txn_context.ttid)))
-                    if compression:
-                        data = decompress(data)
+                    data = decompress_list[compression](data)
                     size = len(data)
                     txn_context.data_size += size
                     if cached:

neo/client/pool.py

@@ -47,7 +47,7 @@ class ConnectionPool(object):
         conn = MTClientConnection(app, app.storage_event_handler, node,
                                   dispatcher=app.dispatcher)
         p = Packets.RequestIdentification(NodeTypes.CLIENT,
-            app.uuid, None, app.name, app.id_timestamp)
+            app.uuid, None, app.name, (), app.id_timestamp)
         try:
             app._ask(conn, p, handler=app.storage_bootstrap_handler)
         except ConnectionClosed:

neo/client/transactions.py

@@ -117,7 +117,7 @@ class Transaction(object):
         if uuid_list:
             return
         del self.data_dict[oid]
-        if type(data) is str:
+        if type(data) is bytes:
             size = len(data)
             self.data_size -= size
             size += self.cache_size

neo/debug.py

@@ -164,3 +164,17 @@ elif IF == 'frames':
         write("Thread %s:\n" % thread_id)
         traceback.print_stack(frame)
     write("End of dump\n")
+
+elif IF == 'profile':
+    DURATION = 60
+    def stop(prof, path):
+        prof.disable()
+        prof.dump_stats(path)
+    @defer
+    def profile(app):
+        import cProfile, threading, time
+        from .lib.protocol import uuid_str
+        path = 'neo-%s-%s.prof' % (uuid_str(app.uuid), time.time())
+        prof = cProfile.Profile()
+        threading.Timer(DURATION, stop, (prof, path)).start()
+        prof.enable()

neo/lib/bootstrap.py

@@ -26,13 +26,14 @@ class BootstrapManager(EventHandler):
     Manage the bootstrap stage, lookup for the primary master then connect to it
     """
 
-    def __init__(self, app, node_type, server=None):
+    def __init__(self, app, node_type, server=None, devpath=()):
         """
         Manage the bootstrap stage of a non-master node, it lookup for the
         primary master node, connect to it then returns when the master node
         is ready.
         """
         self.server = server
+        self.devpath = devpath
         self.node_type = node_type
         self.num_replicas = None
         self.num_partitions = None
@@ -43,7 +44,7 @@ class BootstrapManager(EventHandler):
     def connectionCompleted(self, conn):
         EventHandler.connectionCompleted(self, conn)
         conn.ask(Packets.RequestIdentification(self.node_type, self.uuid,
-            self.server, self.app.name, None))
+            self.server, self.app.name, self.devpath, None))
 
     def connectionFailed(self, conn):
         EventHandler.connectionFailed(self, conn)

neo/lib/compress.py

+#
+# Copyright (C) 2018  Nexedi SA
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 2
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import zlib
+
+decompress_list = (
+    lambda data: data,
+    zlib.decompress,
+)
+
+def parseOption(value):
+    x = value.split('=', 1)
+    try:
+        alg = ('zlib',).index(x[0])
+        if len(x) == 1:
+            return alg, None
+        level = int(x[1])
+    except Exception:
+        raise ValueError("not a valid 'compress' option: %r" % value)
+    if 0 < level <= zlib.Z_BEST_COMPRESSION:
+        return alg, level
+    raise ValueError("invalid compression level: %r" % level)
+
+def getCompress(value):
+    if value:
+        alg, level = (0, None) if value is True else value
+        _compress = zlib.compress
+        if level:
+            zlib_compress = _compress
+            _compress = lambda data: zlib_compress(data, level)
+        alg += 1
+        assert 0 < alg < len(decompress_list), 'invalid compression algorithm'
+        def compress(data):
+            size = len(data)
+            compressed = _compress(data)
+            if len(compressed) < size:
+                return size, alg, compressed
+            return size, 0, data
+        compress._compress = _compress # for testBasicStore
+        return compress
+    return lambda data: (len(data), 0, data)

neo/lib/connector.py

@@ -34,6 +34,7 @@ class SocketConnector(object):
     is_closed = is_server = None
     connect_limit = {}
     CONNECT_LIMIT = 1
+    SOMAXCONN = 5 # for threaded tests
 
     def __new__(cls, addr, s=None):
         if s is None:
@@ -78,7 +79,8 @@ class SocketConnector(object):
     def queue(self, data):
         was_empty = not self.queued
         self.queued += data
-        self.queue_size += len(data)
+        for data in data:
+            self.queue_size += len(data)
         return was_empty
 
     def _error(self, op, exc=None):
@@ -123,7 +125,7 @@ class SocketConnector(object):
         try:
             self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             self._bind(self.addr)
-            self.socket.listen(5)
+            self.socket.listen(self.SOMAXCONN)
         except socket.error, e:
             self.socket.close()
             self._error('listen', e)

neo/lib/exception.py

@@ -26,9 +26,6 @@ class PrimaryFailure(NeoException):
 class StoppedOperation(NeoException):
     pass
 
-class DatabaseFailure(NeoException):
-    pass
-
 class NodeNotReady(NeoException):
     pass
 

neo/lib/interfaces.py

@@ -22,14 +22,13 @@ def check_signature(reference, function):
     a, b, c, d = inspect.getargspec(function)
     x = len(A) - len(a)
     if x < 0: # ignore extra default parameters
-        if x + len(d) < 0:
+        if B or x + len(d) < 0:
             return False
         del a[x:]
         d = d[:x] or None
     elif x: # different signature
-        # We have no need yet to support methods with default parameters.
-        return a == A[:-x] and (b or a and c) and not (d or D)
-    return a == A and b == B and c == C and d == D
+        return a == A[:-x] and (b or a and c) and (d or ()) == (D or ())[:-x]
+    return a == A and (b or not B) and (c or not C) and d == D
 
 def implements(obj, ignore=()):
     ignore = set(ignore)
@@ -55,7 +54,7 @@ def implements(obj, ignore=()):
         while 1:
             name, func = base.pop()
             x = getattr(obj, name)
-            if x.im_class is tobj:
+            if type(getattr(x, '__self__', None)) is tobj:
                 x = x.__func__
                 if x is func:
                     try:

neo/lib/logger.py

@@ -281,3 +281,16 @@ class NEOLogger(Logger):
 logging = NEOLogger()
 signal.signal(signal.SIGRTMIN, lambda signum, frame: logging.flush())
 signal.signal(signal.SIGRTMIN+1, lambda signum, frame: logging.reopen())
+
+def patch():
+    def fork():
+        with logging:
+            pid = os_fork()
+            if not pid:
+                logging._setup()
+        return pid
+    os_fork = os.fork
+    os.fork = fork
+
+patch()
+del patch

neo/lib/node.py

@@ -28,6 +28,7 @@ class Node(object):
 
     _connection = None
     _identified = False
+    devpath = ()
     id_timestamp = None
 
     def __init__(self, manager, address=None, uuid=None, state=NodeStates.DOWN):

neo/lib/patch.py

@@ -25,6 +25,7 @@ def speedupFileStorageTxnLookup():
     from array import array
     from bisect import bisect
     from collections import defaultdict
+    from neo.lib import logging
     from ZODB.FileStorage.FileStorage import FileStorage, FileIterator
 
     typecode = 'L' if array('I').itemsize < 4 else 'I'
@@ -44,6 +45,8 @@ def speedupFileStorageTxnLookup():
             try:
                 index = self._tidindex
             except AttributeError:
+                logging.info("Building index for faster lookup of"
+                             " transactions in the FileStorage DB.")
                 # Cache a sorted list of all the file pos from oid index.
                 # To reduce memory usage, the list is splitted in arrays of
                 # low order 32-bit words.
@@ -52,10 +55,10 @@ def speedupFileStorageTxnLookup():
                     tindex[x >> 32].append(x & 0xffffffff)
                 index = self._tidindex = []
                 for h, l in sorted(tindex.iteritems()):
-                    x = array('I')
-                    x.fromlist(sorted(l))
-                    l = self._read_data_header(h << 32 | x[0])
-                    index.append((l.tid, h, x))
+                    l = array(typecode, sorted(l))
+                    x = self._read_data_header(h << 32 | l[0])
+                    index.append((x.tid, h, l))
+                logging.info("... index built")
             x = bisect(index, (start,)) - 1
             if x >= 0:
                 x, h, index = index[x]

neo/lib/util.py

@@ -15,12 +15,12 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 
-import socket
+import os, socket
 from binascii import a2b_hex, b2a_hex
 from datetime import timedelta, datetime
 from hashlib import sha1
 from Queue import deque
-from struct import pack, unpack
+from struct import pack, unpack, Struct
 from time import gmtime
 
 TID_LOW_OVERFLOW = 2**32
@@ -102,11 +102,10 @@ def addTID(ptid, offset):
         higher = (d.year, d.month, d.day, d.hour, d.minute)
     return packTID(higher, lower)
 
-def u64(s):
-    return unpack('!Q', s)[0]
-
-def p64(n):
-    return pack('!Q', n)
+p64, u64 = (lambda unpack: (
+    unpack.__self__.pack,
+    lambda s: unpack(s)[0]
+))(Struct('!Q').unpack)
 
 def add64(packed, offset):
     """Add a python number to a 64-bits packed value"""
@@ -115,7 +114,7 @@ def add64(packed, offset):
 def dump(s):
     """Dump a binary string in hex."""
     if s is not None:
-        if isinstance(s, str):
+        if isinstance(s, bytes):
             return b2a_hex(s)
         return repr(s)
 
@@ -226,3 +225,25 @@ class cached_property(object):
         if obj is None: return self
         value = obj.__dict__[self.func.__name__] = self.func(obj)
         return value
+
+# This module is always imported before multiprocessing is used, and the
+# main process does not want to change name when task are run in threads.
+spt_pid = os.getpid()
+
+def setproctitle(title):
+    global spt_pid
+    pid = os.getpid()
+    if spt_pid == pid:
+        return
+    spt_pid = pid
+    # Try using https://pypi.org/project/setproctitle/
+    try:
+        # On Linux, this is done by clobbering argv, and the main process
+        # usually has a longer command line than the title of subprocesses.
+        os.environ['SPT_NOENV'] = '1'
+        from setproctitle import setproctitle
+    except ImportError:
+        return
+    finally:
+        del os.environ['SPT_NOENV']
+    setproctitle(title)

neo/master/handlers/identification.py

@@ -24,7 +24,7 @@ from ..app import monotonic_time
 class IdentificationHandler(EventHandler):
 
     def requestIdentification(self, conn, node_type, uuid,
-                              address, name, id_timestamp):
+                              address, name, devpath, id_timestamp):
         app = self.app
         self.checkClusterName(name)
         if address == app.server:
@@ -101,6 +101,8 @@ class IdentificationHandler(EventHandler):
                 uuid=uuid, address=address)
         else:
             node.setUUID(uuid)
+        if devpath:
+            node.devpath = tuple(devpath)
         node.id_timestamp = monotonic_time()
         node.setState(state)
         conn.setHandler(handler)
@@ -120,7 +122,7 @@ class IdentificationHandler(EventHandler):
 class SecondaryIdentificationHandler(EventHandler):
 
     def requestIdentification(self, conn, node_type, uuid,
-                              address, name, id_timestamp):
+                              address, name, devpath, id_timestamp):
         app = self.app
         self.checkClusterName(name)
         if address == app.server:

neo/master/handlers/master.py

@@ -38,7 +38,7 @@ class ElectionHandler(MasterHandler):
             super(ElectionHandler, self).connectionCompleted(conn)
             app = self.app
             conn.ask(Packets.RequestIdentification(NodeTypes.MASTER,
-                app.uuid, app.server, app.name, app.election))
+                app.uuid, app.server, app.name, (), app.election))
 
     def connectionFailed(self, conn):
         super(ElectionHandler, self).connectionFailed(conn)

neo/master/pt.py

@@ -178,7 +178,7 @@ class PartitionTable(neo.lib.pt.PartitionTable):
     def tweak(self, drop_list=()):
         """Optimize partition table
 
-        This reassigns cells in 3 ways:
+        This reassigns cells in 4 ways:
         - Discard cells of nodes listed in 'drop_list'. For partitions with too
           few readable cells, some cells are instead marked as FEEDING. This is
           a preliminary step to drop these nodes, otherwise the partition table
@@ -187,6 +187,8 @@ class PartitionTable(neo.lib.pt.PartitionTable):
         - When a transaction creates new objects (oids are roughly allocated
           sequentially), we expect better performance by maximizing the number
           of involved nodes (i.e. parallelizing writes).
+        - For maximum resiliency, cells of each partition are assigned as far
+          as possible from each other, by checking the topology path of nodes.
 
         Examples of optimal partition tables with np=10, nr=1 and 5 nodes:
 
@@ -215,6 +217,17 @@ class PartitionTable(neo.lib.pt.PartitionTable):
           U.  .U  U.
           .U  U.  U.
           U.  U.  .U
+
+        For the topology, let's consider an example with paths of the form
+        (room, machine, disk):
+        - if there are more rooms than the number of replicas, 2 cells of the
+          same partition must not be assigned in the same room;
+        - otherwise, topology paths are checked at a deeper depth,
+          e.g. not on the same machine and distributed evenly
+               (off by 1) among rooms.
+        But the topology is expected to be optimal, otherwise it is ignored.
+        In some cases, we could fall back to a non-optimal topology but
+        that would cause extra replication if the user wants to fix it.
         """
         # Collect some data in a usable form for the rest of the method.
         node_list = {node: {} for node in self.count_dict
@@ -242,6 +255,67 @@ class PartitionTable(neo.lib.pt.PartitionTable):
                 i += 1
         option_dict = Counter(map(tuple, x))
 
+        # Initialize variables/functions to optimize the topology.
+        devpath_max = []
+        devpaths = [()] * node_count
+        if repeats > 1:
+            _devpaths = [x[0].devpath for x in node_list]
+            max_depth = min(map(len, _devpaths))
+            depth = 0
+            while 1:
+                if depth < max_depth:
+                    depth += 1
+                    x = Counter(x[:depth] for x in _devpaths)
+                    n = len(x)
+                    x = set(x.itervalues())
+                    # TODO: Prove it works. If the code turns out to be:
+                    #       - too pessimistic, the topology is ignored when
+                    #         resiliency could be maximized;
+                    #       - or worse too optimistic, in which case this
+                    #         method raises, possibly after a very long time.
+                    if len(x) == 1 or max(x) * repeats <= node_count:
+                        i, x = divmod(repeats, n)
+                        devpath_max.append((i + 1, x) if x else (i, n))
+                        if n < repeats:
+                            continue
+                        devpaths = [x[:depth] for x in _devpaths]
+                        break
+                logging.warning("Can't maximize resiliency: fix the topology"
+                    " of your storage nodes and make sure they're all running."
+                    " %s storage device failure(s) may be enough to lose all"
+                    " the database." % (repeats - 1))
+                break
+        topology = [{} for _ in xrange(self.np)]
+        def update_topology():
+            for offset in option:
+                n = topology[offset]
+                for i, (j, k) in zip(devpath, devpath_max):
+                    try:
+                        i, x = n[i]
+                    except KeyError:
+                        n[i] = i, x = [0, {}]
+                    if i == j or i + 1 == j and k == sum(
+                          1 for i in n.itervalues() if i[0] == j):
+                        # Too many cells would be assigned at this topology
+                        # node.
+                        return False
+                    n = x
+            # The topology may be optimal with this option. Apply it.
+            for offset in option:
+                n = topology[offset]
+                for i in devpath:
+                    n = n[i]
+                    n[0] += 1
+                    n = n[1]
+            return True
+        def revert_topology():
+            for offset in option:
+                n = topology[offset]
+                for i in devpath:
+                    n = n[i]
+                    n[0] -= 1
+                    n = n[1]
+
         # Strategies to find the "best" permutation of nodes.
         def node_options():
             # The second part of the key goes with the above cosmetic sort.
@@ -291,24 +365,27 @@ class PartitionTable(neo.lib.pt.PartitionTable):
         new = []   # the solution
         stack = [] # data recursion
         def options():
-            return iter(node_options[len(new)][-1])
+            x = node_options[len(new)]
+            return devpaths[x[-2]], iter(x[-1])
         for node_options in node_options(): # for each strategy
-            iter_option = options()
+            devpath, iter_option = options()
             while 1:
                 try:
                     option = next(iter_option)
-                except StopIteration: # 1st strategy only
+                except StopIteration:
                     if new:
-                        iter_option = stack.pop()
-                        option_dict[new.pop()] += 1
+                        devpath, iter_option = stack.pop()
+                        option = new.pop()
+                        revert_topology()
+                        option_dict[option] += 1
                         continue
                     break
-                if option_dict[option]:
+                if option_dict[option] and update_topology():
                     new.append(option)
-                    if len(new) == len(node_list):
+                    if len(new) == node_count:
                         break
-                    stack.append(iter_option)
-                    iter_option = options()
+                    stack.append((devpath, iter_option))
+                    devpath, iter_option = options()
                     option_dict[option] -= 1
             if new:
                 break
@@ -384,13 +461,18 @@ class PartitionTable(neo.lib.pt.PartitionTable):
                 if cell.isReadable():
                     if cell.getNode().isRunning():
                         lost = None
-                    else :
+                    else:
                         cell_list.append(cell)
             for cell in cell_list:
-                if cell.getNode() is not lost:
-                    cell.setState(CellStates.OUT_OF_DATE)
-                    change_list.append((offset, cell.getUUID(),
-                        CellStates.OUT_OF_DATE))
+                node = cell.getNode()
+                if node is not lost:
+                    if cell.isFeeding():
+                        self.removeCell(offset, node)
+                        state = CellStates.DISCARDED
+                    else:
+                        state = CellStates.OUT_OF_DATE
+                        cell.setState(state)
+                    change_list.append((offset, node.getUUID(), state))
         if fully_readable and change_list:
             logging.warning(self._first_outdated_message)
         return change_list

neo/scripts/runner.py

@@ -65,6 +65,7 @@ UNIT_TEST_MODULES = [
     'neo.tests.client.testZODBURI',
     # light functional tests
     'neo.tests.threaded.test',
+    'neo.tests.threaded.testConfig',
     'neo.tests.threaded.testImporter',
     'neo.tests.threaded.testReplication',
     'neo.tests.threaded.testSSL',

neo/storage/app.py

@@ -71,6 +71,7 @@ class Application(BaseApplication):
 
         self.dm.setup(reset=config.getReset(), dedup=config.getDedup())
         self.loadConfiguration()
+        self.devpath = self.dm.getTopologyPath()
 
         # force node uuid from command line argument, for testing purpose only
         if config.getUUID() is not None:
@@ -203,7 +204,8 @@ class Application(BaseApplication):
         pt = self.pt
 
         # search, find, connect and identify to the primary master
-        bootstrap = BootstrapManager(self, NodeTypes.STORAGE, self.server)
+        bootstrap = BootstrapManager(self, NodeTypes.STORAGE, self.server,
+                                     self.devpath)
         self.master_node, self.master_conn, num_partitions, num_replicas = \
             bootstrap.getPrimaryConnection()
         uuid = self.uuid

neo/storage/checker.py

@@ -51,7 +51,7 @@ class Checker(object):
             else:
                 conn = ClientConnection(app, StorageOperationHandler(app), node)
                 conn.ask(Packets.RequestIdentification(NodeTypes.STORAGE,
-                    uuid, app.server, name, app.id_timestamp))
+                    uuid, app.server, name, (), app.id_timestamp))
             self.conn_dict[conn] = node.isIdentified()
         conn_set = set(self.conn_dict)
         conn_set.discard(None)

neo/storage/database/__init__.py

@@ -16,8 +16,6 @@
 
 LOG_QUERIES = False
 
-from neo.lib.exception import DatabaseFailure
-
 DATABASE_MANAGER_DICT = {
     'Importer': 'importer.ImporterDatabaseManager',
     'MySQL': 'mysqldb.MySQLDatabaseManager',
@@ -33,3 +31,6 @@ def getAdapterKlass(name):
 
 def buildDatabaseManager(name, args=(), kw={}):
     return getAdapterKlass(name)(*args, **kw)
+
+class DatabaseFailure(Exception):
+    pass

neo/storage/handlers/client.py

@@ -42,11 +42,11 @@ class ClientOperationHandler(BaseHandler):
         # for read rpc
         return self.app.tm.read_queue
 
-    def askObject(self, conn, oid, serial, tid):
+    def askObject(self, conn, oid, at, before):
         app = self.app
         if app.tm.loadLocked(oid):
             raise DelayEvent
-        o = app.dm.getObject(oid, serial, tid)
+        o = app.dm.getObject(oid, at, before)
         try:
             serial, next_serial, compression, checksum, data, data_serial = o
         except TypeError:

neo/storage/handlers/identification.py

@@ -32,7 +32,7 @@ class IdentificationHandler(EventHandler):
         return self.app.nm
 
     def requestIdentification(self, conn, node_type, uuid, address, name,
-                              id_timestamp):
+                              devpath, id_timestamp):
         self.checkClusterName(name)
         app = self.app
         # reject any incoming connections if not ready

neo/storage/handlers/initialization.py

@@ -28,21 +28,21 @@ class InitializationHandler(BaseMasterHandler):
             raise ProtocolError('Partial partition table received')
         # Install the partition table into the database for persistence.
         cell_list = []
-        offset_list = xrange(pt.getPartitions())
-        unassigned_set = set(offset_list)
-        for offset in offset_list:
+        unassigned = range(pt.getPartitions())
+        for offset in reversed(unassigned):
             for cell in pt.getCellList(offset):
                 cell_list.append((offset, cell.getUUID(), cell.getState()))
                 if cell.getUUID() == app.uuid:
-                    unassigned_set.remove(offset)
+                    unassigned.remove(offset)
         # delete objects database
         dm = app.dm
-        if unassigned_set:
+        if unassigned:
           if app.disable_drop_partitions:
-            logging.info("don't drop data for partitions %r", unassigned_set)
+            logging.info('partitions %r are discarded but actual deletion'
+                         ' of data is disabled', unassigned)
           else:
-            logging.debug('drop data for partitions %r', unassigned_set)
-            dm.dropPartitions(unassigned_set)
+            logging.debug('drop data for partitions %r', unassigned)
+            dm.dropPartitions(unassigned)
 
         dm.changePartitionTable(ptid, cell_list, reset=True)
         dm.commit()
@@ -63,7 +63,7 @@ class InitializationHandler(BaseMasterHandler):
     def askLastIDs(self, conn):
         dm = self.app.dm
         dm.truncate()
-        ltid, _, _, loid = dm.getLastIDs()
+        ltid, loid = dm.getLastIDs()
         conn.answer(Packets.AnswerLastIDs(loid, ltid))
 
     def askPartitionTable(self, conn):
@@ -77,18 +77,10 @@ class InitializationHandler(BaseMasterHandler):
     def validateTransaction(self, conn, ttid, tid):
         dm = self.app.dm
         dm.lockTransaction(tid, ttid)
-        dm.unlockTransaction(tid, ttid)
+        dm.unlockTransaction(tid, ttid, True, True)
         dm.commit()
 
     def startOperation(self, conn, backup):
-        self.app.operational = True
         # XXX: see comment in protocol
-        dm = self.app.dm
-        if backup:
-            if dm.getBackupTID():
-                return
-            tid = dm.getLastIDs()[0] or ZERO_TID
-        else:
-            tid = None
-        dm._setBackupTID(tid)
-        dm.commit()
+        self.app.operational = True
+        self.app.replicator.startOperation(backup)

neo/storage/handlers/master.py

@@ -26,10 +26,7 @@ class MasterOperationHandler(BaseMasterHandler):
     def startOperation(self, conn, backup):
         # XXX: see comment in protocol
         assert self.app.operational and backup
-        dm = self.app.dm
-        if not dm.getBackupTID():
-            dm._setBackupTID(dm.getLastIDs()[0] or ZERO_TID)
-            dm.commit()
+        self.app.replicator.startOperation(backup)
 
     def askLockInformation(self, conn, ttid, tid):
         self.app.tm.lock(ttid, tid)

neo/storage/handlers/storage.py

@@ -75,9 +75,6 @@ class StorageOperationHandler(EventHandler):
                 deleteTransaction(tid)
         assert not pack_tid, "TODO"
         if next_tid:
-            # More than one chunk ? This could be a full replication so avoid
-            # restarting from the beginning by committing now.
-            self.app.dm.commit()
             self.app.replicator.fetchTransactions(next_tid)
         else:
             self.app.replicator.fetchObjects()
@@ -97,15 +94,12 @@ class StorageOperationHandler(EventHandler):
             for serial, oid_list in object_dict.iteritems():
                 for oid in oid_list:
                     deleteObject(oid, serial)
-        # XXX: It should be possible not to commit here if it was the last
-        #      chunk, because we'll either commit again when updating
-        #      'backup_tid' or the partition table.
-        self.app.dm.commit()
         assert not pack_tid, "TODO"
         if next_tid:
             # TODO also provide feedback to master about current replication state (tid)
             self.app.replicator.fetchObjects(next_tid, next_oid)
         else:
+            # This will also commit.
             self.app.replicator.finish()
 
     @checkConnectionIsReplicatorConnection
@@ -267,6 +261,8 @@ class StorageOperationHandler(EventHandler):
                             "partition %u dropped or truncated"
                             % partition), msg_id)
                         return
+                    if not object[2]: # creation undone
+                        object = object[0], 0, ZERO_HASH, '', object[4]
                     # Same as in askFetchTransactions.
                     conn.send(Packets.AddObject(oid, *object), msg_id)
                     yield conn.buffering

neo/storage/transactions.py

@@ -314,12 +314,15 @@ class TransactionManager(EventQueue):
             Unlock transaction
         """
         try:
-            tid = self._transaction_dict[ttid].tid
+            transaction = self._transaction_dict[ttid]
         except KeyError:
             raise ProtocolError("unknown ttid %s" % dump(ttid))
+        tid = transaction.tid
         logging.debug('Unlock TXN %s (ttid=%s)', dump(tid), dump(ttid))
         dm = self._app.dm
-        dm.unlockTransaction(tid, ttid)
+        dm.unlockTransaction(tid, ttid,
+            transaction.voted == 2,
+            transaction.store_dict)
         self._app.em.setTimeout(time() + 1, dm.deferCommit())
         self.abort(ttid, even_if_locked=True)
 
@@ -521,7 +524,6 @@ class TransactionManager(EventQueue):
             assert not even_if_locked
             # See how the master processes AbortTransaction from the client.
             return
-        logging.debug('Abort TXN %s', dump(ttid))
         transaction = self._transaction_dict[ttid]
         locked = transaction.tid
         # if the transaction is locked, ensure we can drop it
@@ -529,6 +531,7 @@ class TransactionManager(EventQueue):
             if not even_if_locked:
                 return
         else:
+            logging.debug('Abort TXN %s', dump(ttid))
             dm = self._app.dm
             dm.abortTransaction(ttid)
             dm.releaseData([x[1] for x in transaction.store_dict.itervalues()],

neo/tests/__init__.py

@@ -28,8 +28,12 @@ import weakref
 import MySQLdb
 import transaction
 
+from ConfigParser import SafeConfigParser
 from cStringIO import StringIO
-from cPickle import Unpickler
+try:
+    from ZODB._compat import Unpickler
+except ImportError:
+    from cPickle import Unpickler
 from functools import wraps
 from inspect import isclass
 from .mock import Mock
@@ -152,8 +156,22 @@ def setupMySQLdb(db_list, user=DB_USER, password='', clear_databases=True):
     conn.commit()
     conn.close()
 
+def ImporterConfigParser(adapter, zodb, **kw):
+    cfg = SafeConfigParser()
+    cfg.add_section("neo")
+    cfg.set("neo", "adapter", adapter)
+    for x in kw.iteritems():
+        cfg.set("neo", *x)
+    for name, zodb in zodb:
+        cfg.add_section(name)
+        for x in zodb.iteritems():
+            cfg.set(name, *x)
+    return cfg
+
 class NeoTestBase(unittest.TestCase):
 
+    maxDiff = None
+
     def setUp(self):
         logging.name = self.setupLog()
         unittest.TestCase.setUp(self)
@@ -172,6 +190,8 @@ class NeoTestBase(unittest.TestCase):
         # Note we don't even abort them because it may require a valid
         # connection to a master node (see Storage.sync()).
         transaction.manager.__init__()
+        if logging._max_size is not None:
+            logging.flush()
 
     class failureException(AssertionError):
         def __init__(self, msg=None):

neo/tests/functional/testClient.py

@@ -202,9 +202,9 @@ class ClientTests(NEOFunctionalTest):
         self.neo.stop()
 
         self.neo = NEOCluster(db_list=['test_neo1'], partitions=3,
-            importer=[("root", {
+            importer={"zodb": [("root", {
                 "storage": "<filestorage>\npath %s\n</filestorage>"
-                            % dfs_storage.getName()})],
+                            % dfs_storage.getName()})]},
             temp_dir=self.getTempDirectory())
         self.neo.start()
         neo_db, neo_conn = self.neo.getZODBConnection()