Overview
| SHA1 Hash: | 48c4e69d2bf23dab30c91609ab5c9b250609b679 |
|---|---|
| Date: | 2007-09-09 17:51:16 |
| User: | drh |
| Comment: | Cluster-based synchronization appears to be working. |
| Timelines: | ancestors | descendants | both | trunk |
| Other Links: | files | ZIP archive | manifest |
Tags And Properties
- branch=trunk inherited from [a28c83647d]
- sym-trunk inherited from [a28c83647d]
Changes
[hide diffs]Modified src/content.c from [52f1a41cd5] to [310f7c7a1f].
@@ -154,10 +154,12 @@
int size;
int rid;
Stmt s1;
Blob cmpr;
Blob hash;
+ int markAsUnclustered = 0;
+
assert( g.repositoryOpen );
if( pBlob && srcId==0 ){
sha1sum_blob(pBlob, &hash);
}else{
blob_init(&hash, zUuid, -1);
@@ -183,10 +185,11 @@
db_end_transaction(0);
return rid;
}
}else{
rid = 0; /* No entry with the same UUID currently exists */
+ markAsUnclustered = 1;
}
db_finalize(&s1);
/* Construct a received-from ID if we do not already have one */
if( g.rcvid==0 && pBlob!=0 ){
@@ -235,14 +238,14 @@
*/
if( srcId ){
db_multi_exec("REPLACE INTO delta(rid,srcid) VALUES(%d,%d)", rid, srcId);
}
- /* Add the element to the unclustered table if it is not a
- ** a phantom
+ /* Add the element to the unclustered table if has never been
+ ** previously seen.
*/
- if( pBlob ){
+ if( markAsUnclustered ){
db_multi_exec("INSERT OR IGNORE INTO unclustered VALUES(%d)", rid);
}
/* Finish the transaction and cleanup */
db_finalize(&s1);
Modified src/schema.c from [7efe24b157] to [2f12edc522].
@@ -162,19 +162,25 @@ @ comment TEXT @ ); @ CREATE INDEX event_i1 ON event(mtime); @ CREATE INDEX event_i2 ON event(objid); @ -@ -- A record of phantoms +@ -- A record of phantoms. A phantom is a record for which we know the +@ -- UUID but we do not (yet) know the file content. @ -- @ CREATE TABLE phantom( @ rid INTEGER PRIMARY KEY -- Record ID of the phantom @ ); @ @ -- Unclustered records. An unclustered record is a record (including @ -- a cluster records themselves) that is not mentioned by some other @ -- cluster. +@ -- +@ -- Phantoms are usually included in the unclustered table. A new cluster +@ -- will never be created that contains a phantom. But another repository +@ -- might send us a cluster that contains entries that are phantoms to +@ -- us. @ -- @ CREATE TABLE unclustered( @ rid INTEGER PRIMARY KEY -- Record ID of the unclustered file @ ); @
Modified src/xfer.c from [bebb266bee] to [00b48b7513].
@@ -322,21 +322,29 @@
db_multi_exec("DELETE FROM unsent");
}
/*
** Check to see if the number of unclustered entries is greater than
-** 100 and if it is, form a new cluster.
+** 100 and if it is, form a new cluster. Unclustered phantoms do not
+** count toward the 100 total. And phantoms are never added to a new
+** cluster.
*/
static void create_cluster(void){
Blob cluster, cksum;
Stmt q;
- int rid;
- if( db_int(0, "SELECT count(*) FROM unclustered")<10 ){
+ int nUncl;
+ nUncl = db_int(0, "SELECT count(*) FROM unclustered"
+ " WHERE NOT EXISTS(SELECT 1 FROM phantom"
+ " WHERE rid=unclustered.rid)");
+ if( nUncl<100 ){
return;
}
blob_zero(&cluster);
- db_prepare(&q, "SELECT uuid FROM unclustered JOIN blob USING(rid)"
+ db_prepare(&q, "SELECT uuid FROM unclustered, blob"
+ " WHERE NOT EXISTS(SELECT 1 FROM phantom"
+ " WHERE rid!=unclustered.rid)"
+ " AND unclustered.rid=blob.rid"
" ORDER BY 1");
while( db_step(&q)==SQLITE_ROW ){
blob_appendf(&cluster, "M %s\n", db_column_text(&q, 0));
}
db_finalize(&q);
@@ -533,10 +541,31 @@
}else{
check_login(&xfer.aToken[1], &xfer.aToken[2], &xfer.aToken[3]);
}
}else
+ /* cookie TEXT
+ **
+ ** A cookie contains a arbitrary-length argument that is server-defined.
+ ** The argument must be encoded so as not to contain any whitespace.
+ ** The server can optionally send a cookie to the client. The client
+ ** might then return the same cookie back to the server on its next
+ ** communication. The cookie might record information that helps
+ ** the server optimize a push or pull.
+ **
+ ** The client is not required to return a cookie. So the server
+ ** must not depend on the cookie. The cookie should be an optimization
+ ** only. The client might also send a cookie that came from a different
+ ** server. So the server must be prepared to distinguish its own cookie
+ ** from cookies originating from other servers. The client might send
+ ** back several different cookies to the server. The server should be
+ ** prepared to sift through the cookies and pick the one that it wants.
+ */
+ if( blob_eq(&xfer.aToken[0], "cookie") && xfer.nToken==2 ){
+ /* Process the cookie */
+ }else
+
/* Unknown message
*/
{
cgi_reset_content();
@ error bad\scommand:\s%F(blob_str(&xfer.line))
@@ -600,10 +629,11 @@
const char *zSCode = db_get("server-code", "x");
const char *zPCode = db_get("project-code", 0);
int nMsg = 0; /* Number of messages sent or received */
int nCycle = 0; /* Number of round trips to the server */
int nFileSend = 0;
+ const char *zCookie; /* Server cookie */
Blob send; /* Text we are sending to the server */
Blob recv; /* Reply we got back from the server */
Xfer xfer; /* Transfer data */
memset(&xfer, 0, sizeof(xfer));
@@ -643,10 +673,18 @@
while( go ){
int newPhantom = 0;
+ /* Send make the most recently received cookie. Let the server
+ ** figure out if this is a cookie that it cares about.
+ */
+ zCookie = db_get("cookie", 0);
+ if( zCookie ){
+ blob_appendf(&send, "cookie %s\n", zCookie);
+ }
+
/* Generate gimme messages for phantoms and leaf messages
** for all leaves.
*/
if( pullFlag ){
request_phantoms(&xfer);
@@ -677,11 +715,10 @@
}
if( pushFlag ){
blob_appendf(&send, "push %s %s\n", zSCode, zPCode);
nMsg++;
}
-
/* Process the reply that came back from the server */
while( blob_line(&recv, &xfer.line) ){
if( blob_buffer(&xfer.line)[0]=='#' ){
continue;
@@ -765,10 +802,23 @@
pullFlag = 1;
blob_appendf(&send, "pull %s %s\n", zSCode, zPCode);
nMsg++;
}else
+ /* cookie TEXT
+ **
+ ** The server might include a cookie in its reply. The client
+ ** should remember this cookie and send it back to the server
+ ** in its next query.
+ **
+ ** Each cookie received overwrites the prior cookie from the
+ ** same server.
+ */
+ if( blob_eq(&xfer.aToken[0], "cookie") && xfer.nToken==2 ){
+ db_set("cookie", blob_str(&xfer.aToken[1]));
+ }else
+
/* error MESSAGE
**
** Report an error
*/
if( blob_eq(&xfer.aToken[0],"error") && xfer.nToken==2 ){
@@ -792,27 +842,25 @@
blob_size(&recv), nMsg,
xfer.nFileRcvd + xfer.nDeltaRcvd + xfer.nDanglingFile,
xfer.nFileRcvd, xfer.nDeltaRcvd, xfer.nDanglingFile);
blob_reset(&recv);
+ nCycle++;
+ go = 0;
+
+ /* If we received one or more files on the previous exchange but
+ ** there are still phantoms, then go another round.
+ */
+ if( (xfer.nFileRcvd+xfer.nDeltaRcvd+xfer.nDanglingFile>0 || newPhantom)
+ && db_exists("SELECT 1 FROM phantom")
+ ){
+ go = 1;
+ }
nMsg = 0;
xfer.nFileRcvd = 0;
xfer.nDeltaRcvd = 0;
xfer.nDanglingFile = 0;
- nCycle++;
- go = 0;
-
- /* If we have received one or more files on this cycle or if
- ** we have received information that has caused us to create
- ** new phantoms and we have one or more phantoms, then go for
- ** another round
- */
- if( (xfer.nFileRcvd+xfer.nDeltaRcvd+xfer.nDanglingFile>0 || newPhantom)
- && db_exists("SELECT 1 FROM phantom")
- ){
- go = 1;
- }
/* If we have one or more files queued to send, then go
** another round
*/
if( xfer.nFileSent+xfer.nDeltaSent>0 ){