lobo/isync
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

split off ephemeral sync record state to a separate member

Browse Source 
this allows us to simplify logging of expiration, as we now can just log
the entire persistent state instead of fiddling with bits.
This commit is contained in:
Oswald Buddenhagen 2017-04-01 17:02:22 +02:00
parent efd72b85cc
commit 4ffe149666
1 changed files with 36 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
src/sync.c
View File

@ -124,20 +124,24 @@ make_flags( int flags, char *buf )
return d; return d;
} }
// These is the (mostly) persistent status of the sync record.
// Most of these bits are actually mutually exclusive. It is a
// bitfield to allow for easy testing for multiple states.
#define S_EXPIRE (1<<0) // the entry is being expired (slave message removal scheduled)
#define S_EXPIRED (1<<1) // the entry is expired (slave message removal confirmed)
#define S_DEAD (1<<7) // ephemeral: the entry was killed and should be ignored
#define S_DEAD (1<<0) /* ephemeral: the entry was killed and should be ignored */ // Ephemeral working set.
#define S_DEL(ms) (1<<(2+(ms))) /* ephemeral: m/s message would be subject to expunge */ #define W_NEXPIRE (1<<0) // temporary: new expiration state
#define S_EXPIRED (1<<4) /* the entry is expired (slave message removal confirmed) */ #define W_DELETE (1<<1) // ephemeral: flags propagation is a deletion
#define S_EXPIRE (1<<5) /* the entry is being expired (slave message removal scheduled) */ #define W_DEL(ms) (1<<(2+(ms))) // ephemeral: m/s message would be subject to expunge
#define S_NEXPIRE (1<<6) /* temporary: new expiration state */
#define S_DELETE (1<<7) /* ephemeral: flags propagation is a deletion */
typedef struct sync_rec { typedef struct sync_rec {
struct sync_rec *next; struct sync_rec *next;
/* string_list_t *keywords; */ /* string_list_t *keywords; */
int uid[2]; /* -2 = pending (use tuid), -1 = skipped (too big), 0 = expired */ int uid[2]; /* -2 = pending (use tuid), -1 = skipped (too big), 0 = expired */
message_t *msg[2]; message_t *msg[2];
uchar status, flags, aflags[2], dflags[2]; uchar status, wstate, flags, aflags[2], dflags[2];
char tuid[TUIDL]; char tuid[TUIDL];
} sync_rec_t; } sync_rec_t;
@ -758,6 +762,7 @@ load_state( sync_vars_t *svars )
srec->status = S_EXPIRE | S_EXPIRED; srec->status = S_EXPIRE | S_EXPIRED;
} else } else
srec->status = 0; srec->status = 0;
srec->wstate = 0;
srec->flags = parse_flags( s ); srec->flags = parse_flags( s );
debug( " entry (%d,%d,%u,%s)\n", srec->uid[M], srec->uid[S], srec->flags, srec->status & S_EXPIRED ? "X" : "" ); debug( " entry (%d,%d,%u,%s)\n", srec->uid[M], srec->uid[S], srec->flags, srec->status & S_EXPIRED ? "X" : "" );
srec->msg[M] = srec->msg[S] = 0; srec->msg[M] = srec->msg[S] = 0;
@ -835,7 +840,7 @@ load_state( sync_vars_t *svars )
(t3 = 0, (sscanf( buf + 2, "%d %d %n", &t1, &t2, &t3 ) < 2) || !t3 || (t - t3 != TUIDL + 2)) : (t3 = 0, (sscanf( buf + 2, "%d %d %n", &t1, &t2, &t3 ) < 2) || !t3 || (t - t3 != TUIDL + 2)) :
c == 'S' || c == '!' ? c == 'S' || c == '!' ?
(sscanf( buf + 2, "%d", &t1 ) != 1) : (sscanf( buf + 2, "%d", &t1 ) != 1) :
c == 'F' || c == 'T' || c == '+' || c == '&' || c == '-' || c == '=' || c == '|' || c == '/' || c == '\\' ? c == 'F' || c == 'T' || c == '+' || c == '&' || c == '-' || c == '=' || c == '|' ?
(sscanf( buf + 2, "%d %d", &t1, &t2 ) != 2) : (sscanf( buf + 2, "%d %d", &t1, &t2 ) != 2) :
(sscanf( buf + 2, "%d %d %d", &t1, &t2, &t3 ) != 3)) (sscanf( buf + 2, "%d %d %d", &t1, &t2, &t3 ) != 3))
{ {
@ -864,6 +869,7 @@ load_state( sync_vars_t *svars )
debug( " new entry(%d,%d)\n", t1, t2 ); debug( " new entry(%d,%d)\n", t1, t2 );
srec->msg[M] = srec->msg[S] = 0; srec->msg[M] = srec->msg[S] = 0;
srec->status = 0; srec->status = 0;
srec->wstate = 0;
srec->flags = 0; srec->flags = 0;
srec->tuid[0] = 0; srec->tuid[0] = 0;
srec->next = 0; srec->next = 0;
@ -915,27 +921,8 @@ load_state( sync_vars_t *svars )
srec->flags = t3; srec->flags = t3;
break; break;
case '~': case '~':
debug( "expire now %d\n", t3 ); debug( "status now %#x\n", t3 );
if (t3) srec->status = t3;
srec->status |= S_EXPIRE;
else
srec->status &= ~S_EXPIRE;
break;
case '\\':
t3 = (srec->status & S_EXPIRED);
debug( "expire back to %d\n", t3 / S_EXPIRED );
if (t3)
srec->status |= S_EXPIRE;
else
srec->status &= ~S_EXPIRE;
break;
case '/':
t3 = (srec->status & S_EXPIRE);
debug( "expired now %d\n", t3 / S_EXPIRE );
if (t3)
srec->status |= S_EXPIRED;
else
srec->status &= ~S_EXPIRED;
break; break;
default: default:
error( "Error: unrecognized journal entry at %s:%d\n", svars->jname, line ); error( "Error: unrecognized journal entry at %s:%d\n", svars->jname, line );
@ -1492,7 +1479,7 @@ box_loaded( int sts, message_t *msgs, int total_msgs, int recent_msgs, void *aux
for (t = 0; t < 2; t++) { for (t = 0; t < 2; t++) {
srec->aflags[t] = srec->dflags[t] = 0; srec->aflags[t] = srec->dflags[t] = 0;
if (srec->msg[t] && (srec->msg[t]->flags & F_DELETED)) if (srec->msg[t] && (srec->msg[t]->flags & F_DELETED))
srec->status |= S_DEL(t); srec->wstate |= W_DEL(t);
if (del[t]) { if (del[t]) {
// The target was newly expunged, so there is nothing to update. // The target was newly expunged, so there is nothing to update.
// The deletion is propagated in the opposite iteration. // The deletion is propagated in the opposite iteration.
@ -1515,7 +1502,7 @@ box_loaded( int sts, message_t *msgs, int total_msgs, int recent_msgs, void *aux
if (svars->chan->ops[t] & OP_DELETE) { if (svars->chan->ops[t] & OP_DELETE) {
debug( " %sing delete\n", str_hl[t] ); debug( " %sing delete\n", str_hl[t] );
srec->aflags[t] = F_DELETED; srec->aflags[t] = F_DELETED;
srec->status |= S_DELETE; srec->wstate |= W_DELETE;
} else { } else {
debug( " not %sing delete\n", str_hl[t] ); debug( " not %sing delete\n", str_hl[t] );
} }
@ -1578,6 +1565,7 @@ box_loaded( int sts, message_t *msgs, int total_msgs, int recent_msgs, void *aux
svars->srecadd = &srec->next; svars->srecadd = &srec->next;
svars->nsrecs++; svars->nsrecs++;
srec->status = 0; srec->status = 0;
srec->wstate = 0;
srec->flags = 0; srec->flags = 0;
srec->tuid[0] = 0; srec->tuid[0] = 0;
srec->uid[1-t] = tmsg->uid; srec->uid[1-t] = tmsg->uid;
@ -1663,7 +1651,7 @@ box_loaded( int sts, message_t *msgs, int total_msgs, int recent_msgs, void *aux
((srec->status & (S_EXPIRE|S_EXPIRED)) == (S_EXPIRE|S_EXPIRED)) || ((srec->status & (S_EXPIRE|S_EXPIRED)) == (S_EXPIRE|S_EXPIRED)) ||
((srec->status & (S_EXPIRE|S_EXPIRED)) && (tmsg->flags & F_DELETED))) { ((srec->status & (S_EXPIRE|S_EXPIRED)) && (tmsg->flags & F_DELETED))) {
/* The message is excess or was already (being) expired. */ /* The message is excess or was already (being) expired. */
srec->status |= S_NEXPIRE; srec->wstate |= W_NEXPIRE;
debug( " old pair(%d,%d) expired\n", srec->uid[M], srec->uid[S] ); debug( " old pair(%d,%d) expired\n", srec->uid[M], srec->uid[S] );
if (svars->mmaxxuid < srec->uid[M]) if (svars->mmaxxuid < srec->uid[M])
svars->mmaxxuid = srec->uid[M]; svars->mmaxxuid = srec->uid[M];
@ -1682,7 +1670,7 @@ box_loaded( int sts, message_t *msgs, int total_msgs, int recent_msgs, void *aux
todel--; todel--;
} else if (todel > 0) { } else if (todel > 0) {
/* The message is excess. */ /* The message is excess. */
srec->status |= S_NEXPIRE; srec->wstate |= W_NEXPIRE;
todel--; todel--;
} }
} }
@ -1703,24 +1691,24 @@ box_loaded( int sts, message_t *msgs, int total_msgs, int recent_msgs, void *aux
if (!srec->tuid[0]) { if (!srec->tuid[0]) {
if (!srec->msg[S]) if (!srec->msg[S])
continue; continue;
uint nex = (srec->status / S_NEXPIRE) & 1; uint nex = (srec->wstate / W_NEXPIRE) & 1;
if (nex != ((srec->status / S_EXPIRED) & 1)) { if (nex != ((srec->status / S_EXPIRED) & 1)) {
/* The record needs a state change ... */ /* The record needs a state change ... */
if (nex != ((srec->status / S_EXPIRE) & 1)) { if (nex != ((srec->status / S_EXPIRE) & 1)) {
/* ... and we need to start a transaction. */ /* ... and we need to start a transaction. */
jFprintf( svars, "~ %d %d %d\n", srec->uid[M], srec->uid[S], nex );
debug( " pair(%d,%d): %d (pre)\n", srec->uid[M], srec->uid[S], nex ); debug( " pair(%d,%d): %d (pre)\n", srec->uid[M], srec->uid[S], nex );
srec->status = (srec->status & ~S_EXPIRE) | (nex * S_EXPIRE); srec->status = (srec->status & ~S_EXPIRE) | (nex * S_EXPIRE);
jFprintf( svars, "~ %d %d %u\n", srec->uid[M], srec->uid[S], srec->status );
} else { } else {
/* ... but the "right" transaction is already pending. */ /* ... but the "right" transaction is already pending. */
debug( " pair(%d,%d): %d (pending)\n", srec->uid[M], srec->uid[S], nex ); debug( " pair(%d,%d): %d (pending)\n", srec->uid[M], srec->uid[S], nex );
} }
} else { } else {
/* Note: the "wrong" transaction may be pending here, /* Note: the "wrong" transaction may be pending here,
* e.g.: S_NEXPIRE = 0, S_EXPIRE = 1, S_EXPIRED = 0. */ * e.g.: W_NEXPIRE = 0, S_EXPIRE = 1, S_EXPIRED = 0. */
} }
} else { } else {
if (srec->status & S_NEXPIRE) { if (srec->wstate & W_NEXPIRE) {
jFprintf( svars, "= %d %d\n", srec->uid[M], srec->uid[S] ); jFprintf( svars, "= %d %d\n", srec->uid[M], srec->uid[S] );
debug( " pair(%d,%d): 1 (abort)\n", srec->uid[M], srec->uid[S] ); debug( " pair(%d,%d): 1 (abort)\n", srec->uid[M], srec->uid[S] );
// If we have so many new messages that some of them are instantly expired, // If we have so many new messages that some of them are instantly expired,
@ -1743,7 +1731,7 @@ box_loaded( int sts, message_t *msgs, int total_msgs, int recent_msgs, void *aux
for (t = 0; t < 2; t++) { for (t = 0; t < 2; t++) {
aflags = srec->aflags[t]; aflags = srec->aflags[t];
dflags = srec->dflags[t]; dflags = srec->dflags[t];
if (srec->status & S_DELETE) { if (srec->wstate & W_DELETE) {
if (!aflags) { if (!aflags) {
/* This deletion propagation goes the other way round. */ /* This deletion propagation goes the other way round. */
continue; continue;
@ -1752,7 +1740,7 @@ box_loaded( int sts, message_t *msgs, int total_msgs, int recent_msgs, void *aux
/* The trigger is an expiration transaction being ongoing ... */ /* The trigger is an expiration transaction being ongoing ... */
if ((t == S) && ((shifted_bit(srec->status, S_EXPIRE, S_EXPIRED) ^ srec->status) & S_EXPIRED)) { if ((t == S) && ((shifted_bit(srec->status, S_EXPIRE, S_EXPIRED) ^ srec->status) & S_EXPIRED)) {
/* ... but the actual action derives from the wanted state. */ /* ... but the actual action derives from the wanted state. */
if (srec->status & S_NEXPIRE) if (srec->wstate & W_NEXPIRE)
aflags |= F_DELETED; aflags |= F_DELETED;
else else
dflags |= F_DELETED; dflags |= F_DELETED;
@ -1949,9 +1937,9 @@ flags_set( int sts, void *aux )
switch (sts) { switch (sts) {
case DRV_OK: case DRV_OK:
if (vars->aflags & F_DELETED) if (vars->aflags & F_DELETED)
vars->srec->status |= S_DEL(t); vars->srec->wstate |= W_DEL(t);
else if (vars->dflags & F_DELETED) else if (vars->dflags & F_DELETED)
vars->srec->status &= ~S_DEL(t); vars->srec->wstate &= ~W_DEL(t);
flags_set_p2( svars, vars->srec, t ); flags_set_p2( svars, vars->srec, t );
break; break;
} }
@ -1965,7 +1953,7 @@ flags_set( int sts, void *aux )
static void static void
flags_set_p2( sync_vars_t *svars, sync_rec_t *srec, int t ) flags_set_p2( sync_vars_t *svars, sync_rec_t *srec, int t )
{ {
if (srec->status & S_DELETE) { if (srec->wstate & W_DELETE) {
debug( " pair(%d,%d): resetting %s UID\n", srec->uid[M], srec->uid[S], str_ms[1-t] ); debug( " pair(%d,%d): resetting %s UID\n", srec->uid[M], srec->uid[S], str_ms[1-t] );
jFprintf( svars, "%c %d %d 0\n", "><"[t], srec->uid[M], srec->uid[S] ); jFprintf( svars, "%c %d %d 0\n", "><"[t], srec->uid[M], srec->uid[S] );
srec->uid[1-t] = 0; srec->uid[1-t] = 0;
@ -1977,15 +1965,15 @@ flags_set_p2( sync_vars_t *svars, sync_rec_t *srec, int t )
jFprintf( svars, "* %d %d %u\n", srec->uid[M], srec->uid[S], nflags ); jFprintf( svars, "* %d %d %u\n", srec->uid[M], srec->uid[S], nflags );
} }
if (t == S) { if (t == S) {
uint nex = (srec->status / S_NEXPIRE) & 1; uint nex = (srec->wstate / W_NEXPIRE) & 1;
if (nex != ((srec->status / S_EXPIRED) & 1)) { if (nex != ((srec->status / S_EXPIRED) & 1)) {
jFprintf( svars, "/ %d %d\n", srec->uid[M], srec->uid[S] );
debug( " pair(%d,%d): expired %d (commit)\n", srec->uid[M], srec->uid[S], nex ); debug( " pair(%d,%d): expired %d (commit)\n", srec->uid[M], srec->uid[S], nex );
srec->status = (srec->status & ~S_EXPIRED) | (nex * S_EXPIRED); srec->status = (srec->status & ~S_EXPIRED) | (nex * S_EXPIRED);
jFprintf( svars, "~ %d %d %u\n", srec->uid[M], srec->uid[S], srec->status );
} else if (nex != ((srec->status / S_EXPIRE) & 1)) { } else if (nex != ((srec->status / S_EXPIRE) & 1)) {
jFprintf( svars, "\\ %d %d\n", srec->uid[M], srec->uid[S] );
debug( " pair(%d,%d): expire %d (cancel)\n", srec->uid[M], srec->uid[S], nex ); debug( " pair(%d,%d): expire %d (cancel)\n", srec->uid[M], srec->uid[S], nex );
srec->status = (srec->status & ~S_EXPIRE) | (nex * S_EXPIRE); srec->status = (srec->status & ~S_EXPIRE) | (nex * S_EXPIRE);
jFprintf( svars, "~ %d %d %u\n", srec->uid[M], srec->uid[S], srec->status );
} }
} }
} }
@ -2151,8 +2139,8 @@ box_closed_p2( sync_vars_t *svars, int t )
for (srec = svars->srecs; srec; srec = srec->next) { for (srec = svars->srecs; srec; srec = srec->next) {
if (srec->status & S_DEAD) if (srec->status & S_DEAD)
continue; continue;
if (srec->uid[S] <= 0 || ((srec->status & S_DEL(S)) && (svars->state[S] & ST_DID_EXPUNGE))) { if (srec->uid[S] <= 0 || ((srec->wstate & W_DEL(S)) && (svars->state[S] & ST_DID_EXPUNGE))) {
if (srec->uid[M] <= 0 || ((srec->status & S_DEL(M)) && (svars->state[M] & ST_DID_EXPUNGE)) || if (srec->uid[M] <= 0 || ((srec->wstate & W_DEL(M)) && (svars->state[M] & ST_DID_EXPUNGE)) ||
((srec->status & S_EXPIRED) && svars->maxuid[M] >= srec->uid[M] && svars->mmaxxuid >= srec->uid[M])) { ((srec->status & S_EXPIRED) && svars->maxuid[M] >= srec->uid[M] && svars->mmaxxuid >= srec->uid[M])) {
debug( " -> killing (%d,%d)\n", srec->uid[M], srec->uid[S] ); debug( " -> killing (%d,%d)\n", srec->uid[M], srec->uid[S] );
srec->status = S_DEAD; srec->status = S_DEAD;
@ -2162,7 +2150,7 @@ box_closed_p2( sync_vars_t *svars, int t )
jFprintf( svars, "> %d %d 0\n", srec->uid[M], srec->uid[S] ); jFprintf( svars, "> %d %d 0\n", srec->uid[M], srec->uid[S] );
srec->uid[S] = 0; srec->uid[S] = 0;
} }
} else if (srec->uid[M] > 0 && ((srec->status & S_DEL(M)) && (svars->state[M] & ST_DID_EXPUNGE))) { } else if (srec->uid[M] > 0 && ((srec->wstate & W_DEL(M)) && (svars->state[M] & ST_DID_EXPUNGE))) {
debug( " -> orphaning ([%d],%d)\n", srec->uid[M], srec->uid[S] ); debug( " -> orphaning ([%d],%d)\n", srec->uid[M], srec->uid[S] );
jFprintf( svars, "< %d %d 0\n", srec->uid[M], srec->uid[S] ); jFprintf( svars, "< %d %d 0\n", srec->uid[M], srec->uid[S] );
srec->uid[M] = 0; srec->uid[M] = 0;