diff options
| -rw-r--r-- | docs/design/008-module-format.md | 25 | ||||
| -rw-r--r-- | prototypes/connections/module-list.c | 6 | ||||
| -rw-r--r-- | prototypes/connections/module-list.h | 3 | ||||
| -rw-r--r-- | prototypes/connections/module-save.c | 17 | ||||
| -rw-r--r-- | prototypes/connections/module-save.h | 7 | ||||
| -rw-r--r-- | prototypes/connections/module-store.c | 22 | ||||
| -rw-r--r-- | prototypes/connections/module.c | 188 | ||||
| -rw-r--r-- | prototypes/connections/module.h | 41 | ||||
| -rw-r--r-- | prototypes/connections/pure-run.c | 17 | ||||
| -rw-r--r-- | prototypes/connections/test-module.c | 141 |
10 files changed, 361 insertions, 106 deletions
diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md index 1e27976..4c5a9ac 100644 --- a/docs/design/008-module-format.md +++ b/docs/design/008-module-format.md @@ -125,7 +125,7 @@ v5는 값을 가진 채 합류하는 분기(조건식)와, 리스트 항목마� ## 실행 -`PureApp`은 모듈 참조, 상태 값, text 상태 draft, 리스트, 열린 이벤트 하나를 가진다(1,808 B, `PureModule`은 상수 +`PureApp`은 모듈 참조, 상태 값, text 상태 draft, 리스트, 열린 이벤트 하나를 가진다(4,400 B, `PureModule`은 상수 블롭을 포함해 28,600 B). text 상태는 저장소 파일이 없는 draft라 1 MiB 텍스트도 조각 트리로 편집한다. 커서는 바이트 위치이며 항상 글자 경계에 있다. 넣기는 그 위치 이후의 커서를 뒤로(삽입 위치의 커서 포함) 밀고, 지우기는 범위 안의 커서를 범위 처음으로 모은다. 모든 이벤트에서 text draft를 잡아 두고, 실패하면 이벤트 전 버전으로 돌린다. @@ -173,19 +173,28 @@ text·리스트가 남을 때만 위치를 유지한다. 이름·종류가 같은 필드를 옮기고 새 필드는 기본값을 쓴 뒤 다시 저장한다(교체와 같은 규칙, 마이그레이션 스크립트 없음). 커서가 text 범위나 글자 경계를 벗어나거나, 인덱스가 리스트 밖이면 0이 된다. 깨진 리스트·text(형식 오류, UTF-8 아님)는 시작을 `store_error`로 멈춘다. 조용히 버리지 않는다. -- **이벤트가 곧 커밋이다.** 키·text·메시지 이벤트가 성공하면, 바뀐 상태 문서들과 메시지의 문서를 저장소의 - **한 커밋**(`pure_draft_commit_all`)으로 공개한다. 모두 보이거나 하나도 보이지 않는다. 커밋이 실패하면 이벤트가 - 실패하고 메모리의 상태도 되돌린다. 결과 불명(`uncertain`)이나 다른 기록자와의 충돌 뒤에는 앱이 저장소와 어긋나 - 이후 이벤트가 충돌로 실패한다. 다시 시작하면 저장소에 있는 상태로 이어진다. -- **frame은 늦게 쓴다.** 초당 60번 오는 frame 이벤트는 메모리에서 원자적이고, 바뀐 것은 다음 이벤트의 커밋이나 - `pure_app_save`에 함께 들어간다. `pure_app_stop`만 하면 저장하지 않은 frame 변경은 사라진다. +- **저장은 묶음이다.** 저장 한 번이 마지막 저장 이후 메시지가 고친 문서들과 바뀐 상태 문서들을 저장소의 + **한 커밋**(`pure_draft_commit_all`)으로 공개한다. 모두 보이거나 하나도 보이지 않는다. 저장이 실패하면 마지막 저장 + 이후의 이벤트가 **모두** 되돌려진다(값, text와 리스트 draft, 리스트 되돌리기 기록, 이벤트·tick 수). 그래서 + "성공을 알린 이벤트는 저장됐고, 실패를 알린 이벤트는 흔적이 없다"가 성립한다. 결과 불명(`uncertain`)이나 다른 + 기록자와의 충돌 뒤에는 앱이 저장소와 어긋나 이후 저장이 충돌로 실패한다. 다시 시작하면 저장소의 상태로 이어진다. +- **기본은 이벤트마다 저장.** 키·text·메시지 이벤트는 저장까지 마치고 돌아온다(묶음 크기 1). 초당 60번 오는 + frame 이벤트는 다음 이벤트나 `pure_app_save`에 함께 들어간다. `pure_app_stop`만 하면 저장하지 않은 변경은 사라진다. +- **지연 저장(그룹 커밋).** `pure_app_defer(app, true)`면 이벤트는 바로 적용되고, 호스트가 입력을 몰아 처리한 뒤 + `pure_app_save` 한 번으로 저장한다. 이벤트는 그 뒤 저장이 성공해야 확정이고, 그 전에는 응답을 내보내지 않는 + 것이 호스트의 몫이다. 한 묶음에서 키 하나의 문서는 한 번만 열린다. 같은 키의 두 번째 메시지나 16개 키를 넘는 + 메시지는 먼저 묶음을 저장하므로 문서 리비전은 저장마다 한 칸씩 움직인다(응답에 알린 리비전 = 저장된 리비전). + Pi 5의 SD(ext4)에서 이벤트 600개: 이벤트마다 저장 6,282 ms(10.47 ms/이벤트), 8개씩 973 ms, 64개씩 341 ms, + 한 번에 194 ms. `pure-run --store 경로 --batch N`으로 잰다. - **교체.** 모듈을 바꾸면 새 상태는 저장된 값이 있으면 그것으로 시작하고, 버린 상태의 저장본은 그대로 남는다. 새 모듈의 상태는 다음 이벤트나 저장에서 공개된다. 검증(`test-module`): 같은 무작위 이벤트열을 메모리 앱과 수시로 재시작하는 저장 앱에 주어 매 단계 상태가 비트 단위로 같다(리스트 앱·text 앱 각 400단계, 재시작 20회 이상). 수·text·리스트를 한꺼번에 바꾸는 이벤트 중 커밋의 모든 단계에서 프로세스를 죽여도 다시 시작하면 전부 이전 아니면 전부 이후다. 메시지의 문서와 상태가 함께 커밋되고 함께 실패한다. -레코드 변경은 재시작과 교체 모두에서 항목을 옮긴다. 비용(Pi 5, SD의 ext4): 저장하는 이벤트 하나에 약 10.6 ms가 +레코드 변경은 재시작과 교체 모두에서 항목을 옮긴다. 지연 저장에서는 저장된 이벤트만 다시 재생한 메모리 앱과, +무작위로 저장하고 가끔 저장 없이 멈췄다 다시 시작하는 저장 앱이 매 단계 같다(리스트·text 앱 각 600단계). 묶음 저장이 +실패하면 묶음 전체가 되돌려지고, 저장 중 프로세스가 죽으면 묶음 전체가 남거나 전혀 남지 않는다. 비용(Pi 5, SD의 ext4): 저장하는 이벤트 하나에 약 10.6 ms가 더 든다. 대부분 두 번의 `fsync`다. 300항목 리스트에서 인덱스 이동 한 번은 파일을 약 220 B 키운다(압축 전). 아직 런타임 코어(`runtime.c`)의 작업·예산 모델과 연결하지 않았다. 호스트가 직접 호출한다. 실행기 코드는 diff --git a/prototypes/connections/module-list.c b/prototypes/connections/module-list.c index ba67143..374bc78 100644 --- a/prototypes/connections/module-list.c +++ b/prototypes/connections/module-list.c @@ -234,9 +234,9 @@ PureCell pure_list_get(const PureList *list, size_t at, unsigned field) { return list->items[at][field]; } -void pure_undo_rollback(PureUndo *undo) { +void pure_undo_rollback(PureUndo *undo, size_t mark) { Entry *entries = undo->entries; - for (size_t i = undo->count; i-- > 0;) { + for (size_t i = undo->count; i-- > mark;) { Entry *e = &entries[i]; if (e->change == INSERTED) free_item(e->list, take(e->list, e->at)); else if (e->change == REMOVED) place(e->list, e->at, e->item); /* capacity is still there */ @@ -245,7 +245,7 @@ void pure_undo_rollback(PureUndo *undo) { e->list->items[e->at][e->field] = e->old; } } - undo->count = 0; + undo->count = mark; } void pure_undo_commit(PureUndo *undo) { diff --git a/prototypes/connections/module-list.h b/prototypes/connections/module-list.h index e79ee46..b568a05 100644 --- a/prototypes/connections/module-list.h +++ b/prototypes/connections/module-list.h @@ -35,7 +35,8 @@ void pure_list_mirror(PureList *, PureDraft *, size_t header); PureResult pure_list_write(PureList *); /* Append the items stored in bytes; each field is checked. Before mirroring. */ PureResult pure_list_read(PureList *, const unsigned char *bytes, size_t size); -void pure_undo_rollback(PureUndo *); +/* Undo the changes logged after mark (a count taken earlier), newest first. */ +void pure_undo_rollback(PureUndo *, size_t mark); void pure_undo_commit(PureUndo *); void pure_undo_free(PureUndo *); #endif diff --git a/prototypes/connections/module-save.c b/prototypes/connections/module-save.c index 40b3934..573ac3f 100644 --- a/prototypes/connections/module-save.c +++ b/prototypes/connections/module-save.c @@ -244,9 +244,15 @@ PureResult pure_save_open_values(PureApp *app, const PureModule *m, double *valu static bool same_version(PureVersion a, PureVersion b) { return a.ref == b.ref && a.bytes == b.bytes; } -PureResult pure_save_commit(PureApp *app, const double *values, void *document, int64_t *revision) { - if (!pure_save_enabled(app)) - return document ? app->data.commit(app->data.context, document, revision) : PURE_OK; +PureResult pure_save_commit(PureApp *app, const double *values, void *const *documents, size_t document_count, + int64_t *revisions) { + if (!pure_save_enabled(app)) { + /* Without a host that saves together, a save has one document at most. */ + PureResult result = PURE_OK; + for (size_t i = 0; i < document_count && result == PURE_OK; ++i) + result = app->data.commit(app->data.context, documents[i], &revisions[i]); + return result; + } const PureModule *m = app->module; PureDraft *drafts[2 * PURE_MODULE_STATES + 1]; uint32_t states[2 * PURE_MODULE_STATES + 1]; @@ -278,9 +284,8 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *document, drafts[count++] = app->stored_values; } } - if (!count && !document) return PURE_OK; - int64_t ignored; - PureResult result = app->data.save(app->data.context, document, drafts, count, document ? revision : &ignored); + if (!count && !document_count) return PURE_OK; + PureResult result = app->data.save(app->data.context, documents, document_count, drafts, count, revisions); for (size_t k = 0; result == PURE_OK && k < count; ++k) if (states[k] != UINT32_MAX) app->saved[states[k]] = pure_draft_version(drafts[k]); if (values_changed) { diff --git a/prototypes/connections/module-save.h b/prototypes/connections/module-save.h index 90e0701..b6db997 100644 --- a/prototypes/connections/module-save.h +++ b/prototypes/connections/module-save.h @@ -19,9 +19,10 @@ PureResult pure_save_open_list(PureApp *, const PureModule *m, uint32_t state, P PureResult pure_save_open_values(PureApp *, const PureModule *m, double *values, bool load); /* Store a list whose record changed in its draft, header and items anew. */ PureResult pure_save_rewrite_list(const PureModule *m, uint32_t state, PureList *list, PureDraft *draft); -/* Publish every state changed since the last save (values are the pending - * numbers), with document when not NULL; revision is the document's. */ -PureResult pure_save_commit(PureApp *, const double *values, void *document, int64_t *revision); +/* Publish the documents and every state changed since the last save + * (values are the states' numbers); revisions are the documents'. */ +PureResult pure_save_commit(PureApp *, const double *values, void *const *documents, size_t document_count, + int64_t *revisions); /* A list with record's fields, taking from `from` each field of the same name * and kind and giving the others their defaults (module.c). */ diff --git a/prototypes/connections/module-store.c b/prototypes/connections/module-store.c index 6335903..02d6361 100644 --- a/prototypes/connections/module-store.c +++ b/prototypes/connections/module-store.c @@ -42,17 +42,19 @@ static PureResult state(void *context, const char *name, PureDraft **out) { return pure_state_open(context, name, out); } -/* The document and the app's changed states in one tree commit. */ -static PureResult save(void *context, void *document, PureDraft *const *states, size_t count, int64_t *revision) { +enum { SAVED = PURE_APP_PENDING + 2 * PURE_MODULE_STATES + 1 }; + +/* The documents and the app's changed states in one tree commit. */ +static PureResult save(void *context, void *const *documents, size_t document_count, PureDraft *const *states, + size_t count, int64_t *revisions) { (void)context; - PureDraft *drafts[2 * PURE_MODULE_STATES + 2] = {0}; - int64_t revisions[2 * PURE_MODULE_STATES + 2] = {0}; - size_t n = 0; - if (count > 2 * PURE_MODULE_STATES + 1) return PURE_LIMIT; - if (document) drafts[n++] = document; - for (size_t i = 0; i < count; ++i) drafts[n++] = states[i]; - PureResult result = pure_draft_commit_all(drafts, n, revisions); - if (document) *revision = revisions[0]; + PureDraft *drafts[SAVED] = {0}; + int64_t all[SAVED] = {0}; + if (document_count > PURE_APP_PENDING || count > SAVED - document_count) return PURE_LIMIT; + for (size_t i = 0; i < document_count; ++i) drafts[i] = documents[i]; + for (size_t i = 0; i < count; ++i) drafts[document_count + i] = states[i]; + PureResult result = pure_draft_commit_all(drafts, document_count + count, all); + for (size_t i = 0; i < document_count; ++i) revisions[i] = all[i]; return result; } diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c index 362919b..62f00c7 100644 --- a/prototypes/connections/module.c +++ b/prototypes/connections/module.c @@ -1075,13 +1075,14 @@ PureResult pure_app_start(PureApp *app, const PureModule *module, PureData data) return result; } -PureResult pure_app_save(PureApp *app) { - int64_t revision; - return pure_save_commit(app, app->values, NULL, &revision); +static bool pending(const PureApp *app, const void *document) { + for (unsigned i = 0; i < app->pending_count; ++i) if (app->pending[i].document == document) return true; + return false; } void pure_app_finish(PureApp *app) { - if (app->document) app->data.close(app->data.context, app->document); + /* A document in the unsaved burst stays open until the save. */ + if (app->document && !pending(app, app->document)) app->data.close(app->data.context, app->document); free(app->copied); app->document = NULL; app->copied = NULL; @@ -1090,8 +1091,75 @@ void pure_app_finish(PureApp *app) { app->arena.used = 0; } +/* The start of a burst: what a failed save returns to. Text and stored list + * drafts stay held until the save, so every version since stays readable. */ +static void open_burst(PureApp *app) { + if (app->open) return; + const PureModule *m = app->module; + memcpy(app->durable_values, app->values, sizeof(app->values)); + app->durable_ticks = app->ticks; + app->durable_events = app->events; + for (uint32_t i = 0; i < m->states; ++i) { + if (app->texts[i]) { + pure_draft_hold(app->texts[i], true); + app->durable_texts[i] = pure_draft_version(app->texts[i]); + } + if (app->list_drafts[i]) { + pure_draft_hold(app->list_drafts[i], true); + app->durable_lists[i] = pure_draft_version(app->list_drafts[i]); + } + } + app->open = true; +} + +/* End a burst, keeping it (saved) or undoing all of it. The documents it held + * close, except the current event's, which pure_app_finish closes. */ +static void close_burst(PureApp *app, bool keep) { + for (unsigned i = 0; i < app->pending_count; ++i) + if (app->pending[i].document != app->document) app->data.close(app->data.context, app->pending[i].document); + app->pending_count = 0; + if (!app->open) return; + const PureModule *m = app->module; + for (uint32_t i = 0; i < m->states; ++i) { + if (app->texts[i]) { + if (!keep) pure_draft_restore(app->texts[i], app->durable_texts[i]); + pure_draft_hold(app->texts[i], false); + } + if (app->list_drafts[i]) { + if (!keep) pure_draft_restore(app->list_drafts[i], app->durable_lists[i]); + pure_draft_hold(app->list_drafts[i], false); + } + } + if (keep) pure_undo_commit(&app->undo); + else { + pure_undo_rollback(&app->undo, 0); + memcpy(app->values, app->durable_values, sizeof(app->values)); + app->ticks = app->durable_ticks; + app->events = app->durable_events; + } + app->open = false; +} + +PureResult pure_app_save(PureApp *app) { + void *documents[PURE_APP_PENDING]; + int64_t revisions[PURE_APP_PENDING]; + size_t count = 0; + for (unsigned i = 0; i < app->pending_count; ++i) + if (app->pending[i].edited) documents[count++] = app->pending[i].document; + PureResult result = pure_save_commit(app, app->values, documents, count, revisions); + close_burst(app, result == PURE_OK); + return result; +} + +bool pure_app_defer(PureApp *app, bool deferred) { + if (deferred && !pure_save_enabled(app)) return false; + app->deferred = deferred; + return true; +} + void pure_app_stop(PureApp *app) { pure_app_finish(app); + close_burst(app, true); /* unsaved changes are dropped with the app */ for (uint32_t i = 0; i < PURE_MODULE_STATES; ++i) { pure_draft_close(app->texts[i]); pure_list_free(app->lists[i]); @@ -1104,38 +1172,26 @@ void pure_app_stop(PureApp *app) { memset(app, 0, sizeof(*app)); } -/* Every text draft and stored list draft is held for the event; on failure - * each returns to the version it had when the event began. */ -typedef struct { PureVersion before[PURE_MODULE_STATES], lists[PURE_MODULE_STATES]; } Texts; +/* Where one event began inside the burst: a failed event returns there. */ +typedef struct { PureVersion texts[PURE_MODULE_STATES], lists[PURE_MODULE_STATES]; size_t undo; bool opened; } Mark; -static void hold_texts(PureApp *app, Texts *texts) { +static void mark_event(PureApp *app, Mark *mark) { + mark->opened = !app->open; + open_burst(app); for (uint32_t i = 0; i < app->module->states; ++i) { - if (app->texts[i]) { - pure_draft_hold(app->texts[i], true); - texts->before[i] = pure_draft_version(app->texts[i]); - } - if (app->list_drafts[i]) { - pure_draft_hold(app->list_drafts[i], true); - texts->lists[i] = pure_draft_version(app->list_drafts[i]); - } + if (app->texts[i]) mark->texts[i] = pure_draft_version(app->texts[i]); + if (app->list_drafts[i]) mark->lists[i] = pure_draft_version(app->list_drafts[i]); } + mark->undo = app->undo.count; } -/* Texts and stored lists return to their versions and lists undo their log, - * or all keep what the event did. */ -static void settle(PureApp *app, const Texts *texts, bool keep) { +static void undo_event(PureApp *app, const Mark *mark) { for (uint32_t i = 0; i < app->module->states; ++i) { - if (app->texts[i]) { - if (!keep) pure_draft_restore(app->texts[i], texts->before[i]); - pure_draft_hold(app->texts[i], false); - } - if (app->list_drafts[i]) { - if (!keep) pure_draft_restore(app->list_drafts[i], texts->lists[i]); - pure_draft_hold(app->list_drafts[i], false); - } + if (app->texts[i]) pure_draft_restore(app->texts[i], mark->texts[i]); + if (app->list_drafts[i]) pure_draft_restore(app->list_drafts[i], mark->lists[i]); } - if (keep) pure_undo_commit(&app->undo); - else pure_undo_rollback(&app->undo); + pure_undo_rollback(&app->undo, mark->undo); + if (mark->opened) close_burst(app, true); /* nothing happened: release the holds */ } /* All handlers of one event run in module order against pending states. */ @@ -1152,29 +1208,24 @@ static PureResult run_handlers(PureApp *app, uint32_t event, Event *ev, double * return PURE_OK; } -/* A key or text event stores what it changed before it counts; a frame - * leaves that to the next event (frames come 60 times a second). */ +/* A key or text event saves before it returns unless saving is deferred; a + * frame leaves that to the next event (frames come 60 times a second). */ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) { double pending[PURE_MODULE_STATES]; - Texts texts; - hold_texts(app, &texts); + Mark mark; + mark_event(app, &mark); PureResult result = run_handlers(app, event, ev, pending); - int64_t revision; - if (result == PURE_OK && event) result = pure_save_commit(app, pending, NULL, &revision); - settle(app, &texts, result == PURE_OK); - if (result == PURE_OK) { - memcpy(app->values, pending, sizeof(pending)); - ++app->events; - } - return result; + if (result != PURE_OK) { undo_event(app, &mark); return result; } + memcpy(app->values, pending, sizeof(pending)); + ++app->events; + if (!event) ++app->ticks; + return event && !app->deferred ? pure_app_save(app) : PURE_OK; } PureResult pure_app_frame(PureApp *app) { pure_app_finish(app); Event ev = {.app = app, .arena = &app->arena, .dt = 1.0 / 60.0}; - PureResult result = dispatch(app, 0, &ev); - if (result == PURE_OK) ++app->ticks; - return result; + return dispatch(app, 0, &ev); } PureResult pure_app_input(PureApp *app, uint32_t input, PureEdge edge) { @@ -1220,32 +1271,47 @@ PureResult pure_app_message(PureApp *app, uint32_t input, const char *key, int64 return PURE_INVALID; const PureInputShape *shape = &m->input[input]; if (payload.size < shape->prefix) return fault(app, PURE_FAULT_LIMIT, UINT32_MAX); - app->payload = payload; if (shape->keyed && m->collections) { if (!app->data.open) return PURE_INVALID; + /* A key joins a burst once, so its revision moves once per save. */ + bool seen = app->pending_count == PURE_APP_PENDING; + for (unsigned i = 0; i < app->pending_count && !seen; ++i) seen = !strcmp(app->pending[i].key, key); + if (seen) { + PureResult saved = pure_app_save(app); + if (saved != PURE_OK) return saved; + } PureResult opened = app->data.open(app->data.context, 0, key, &app->document, &app->revision); - if (opened != PURE_OK) { app->document = NULL; pure_app_finish(app); return opened; } + if (opened != PURE_OK) { app->document = NULL; return opened; } if (base >= 0 && base != app->revision) { pure_app_finish(app); return PURE_CONFLICT; } } + app->payload = payload; Event ev = {.app = app, .shape = shape, .arena = &app->arena, .exists = app->revision > 0}; - double pending[PURE_MODULE_STATES]; - Texts texts; - hold_texts(app, &texts); - PureResult result = run_handlers(app, 1 + 2 * input, &ev, pending); + double pending_values[PURE_MODULE_STATES]; + Mark mark; + mark_event(app, &mark); + PureResult result = run_handlers(app, 1 + 2 * input, &ev, pending_values); if (result == PURE_OK && ev.modified && base < 0) result = PURE_INVALID; - /* Commit rebases the working version and releasing texts may reclaim + /* A save rebases an edited document and releasing texts may reclaim * earlier text versions: keep such a reply by copying it first. */ if (result == PURE_OK && app->has_reply && ((app->reply.source == FROM_DOCUMENT && ev.modified) || app->reply.source == FROM_TEXT)) result = keep_reply(app); - if (result == PURE_OK) - result = pure_save_commit(app, pending, ev.modified ? app->document : NULL, &app->revision); - /* States, texts and lists change only with a successful commit (or no document edit). */ - settle(app, &texts, result == PURE_OK); - if (result == PURE_OK) { - memcpy(app->values, pending, sizeof(pending)); - ++app->events; - } else if (result != PURE_UNCERTAIN) pure_app_finish(app); + if (result != PURE_OK) { + undo_event(app, &mark); + pure_app_finish(app); + return result; + } + memcpy(app->values, pending_values, sizeof(pending_values)); + ++app->events; + if (app->document) { + app->pending[app->pending_count].document = app->document; + app->pending[app->pending_count].edited = ev.modified; + memcpy(app->pending[app->pending_count++].key, key, strlen(key) + 1); + app->revision += ev.modified; /* a save moves an edited document one revision */ + } + if (app->deferred) return PURE_OK; + result = pure_app_save(app); + if (result != PURE_OK) pure_app_finish(app); return result; } @@ -1359,6 +1425,10 @@ static PureResult keep_list(PureList **out, PureList *from, const PureModule *ol * a state the new module drops stays stored as it was. */ PureResult pure_app_replace(PureApp *app, const PureModule *module) { pure_app_finish(app); + if (app->open || app->pending_count) { + PureResult saved = pure_app_save(app); /* the burst belongs to the module that made it */ + if (saved != PURE_OK) return saved; + } const PureModule *old = app->module; bool stored = pure_save_enabled(app); double values[PURE_MODULE_STATES] = {0}; diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h index d2d4b7a..c5d4f52 100644 --- a/prototypes/connections/module.h +++ b/prototypes/connections/module.h @@ -19,7 +19,7 @@ enum { PURE_MODULE_NAME = 32, PURE_MODULE_STRINGS = 32768, PURE_MODULE_HEADER = 40, PURE_MODULE_FIELDS = 32, PURE_MODULE_COLLECTIONS = 1, PURE_MODULE_CONSTANTS = 64, PURE_MODULE_CONSTANT_BYTES = 8192, PURE_MODULE_RECORDS = 8, PURE_MODULE_RECORD_FIELDS = 64, - PURE_RECORD_FIELDS = 16, PURE_MODULE_JOINS = 8 + PURE_RECORD_FIELDS = 16, PURE_MODULE_JOINS = 8, PURE_APP_PENDING = 16 }; typedef enum { @@ -123,10 +123,11 @@ typedef struct { void (*close)(void *context, void *document); /* Optional: the host keeps the app's states. state opens a runtime-owned * value by name as a draft (a missing one is empty at revision 0); save - * publishes the given state drafts together with document (or NULL), - * all or nothing, and names the document's new revision. */ + * publishes the given documents and state drafts together, all or + * nothing, and names each document's new revision. */ PureResult (*state)(void *context, const char *name, PureDraft **out); - PureResult (*save)(void *context, void *document, PureDraft *const *states, size_t count, int64_t *revision); + PureResult (*save)(void *context, void *const *documents, size_t document_count, PureDraft *const *states, + size_t count, int64_t *revisions); } PureData; /* A byte or text value inside one event or view: a range of the payload, a @@ -138,10 +139,15 @@ typedef struct { unsigned char *bytes; size_t used, capacity; } PureArena; * runtime-owned drafts; cursors are positions the runtime keeps on character * boundaries; lists are runtime-owned and an index always names an item * (or 0 in an empty list). Every event is atomic: on failure nothing changes. - * When the data host keeps states, the app starts from the stored ones and - * every key, text and message event stores what it changed with its own - * commit (with the message's document); frame events are stored with the - * next one, or by pure_app_save. */ + * + * Saving. When the data host keeps states the app starts from the stored + * ones. A save publishes, in one commit, the documents edited by messages and + * the states changed since the last save; if it fails, every event since the + * last save is undone. By default each key, text and message event saves + * before it returns (a frame waits for the next one). Deferred, events apply + * at once and a host saves a burst with one commit: an event is durable once + * a later save succeeds, and each key's document joins a burst once (a second + * message for it saves the burst first). */ typedef struct { const PureModule *module; PureData data; @@ -152,6 +158,14 @@ typedef struct { PureDraft *list_drafts[PURE_MODULE_STATES]; /* stored lists, when kept */ PureDraft *stored_values; /* numbers, cursors and indices, when kept */ PureVersion saved[PURE_MODULE_STATES]; /* text and list versions at the last save */ + /* Events since the last save: what a failed save returns to, and the + * documents messages opened (held until the save). */ + bool deferred, open; + double durable_values[PURE_MODULE_STATES]; + PureVersion durable_texts[PURE_MODULE_STATES], durable_lists[PURE_MODULE_STATES]; + uint64_t durable_ticks, durable_events; + struct { void *document; char key[65]; bool edited; } pending[PURE_APP_PENDING]; + unsigned pending_count; PureFault fault; uint32_t fault_pc; uint64_t ticks, events; @@ -166,8 +180,11 @@ typedef struct { } PureApp; PureResult pure_app_start(PureApp *, const PureModule *, PureData data); -/* Store every state changed since the last save (after frame events). */ +/* Publish every change since the last save in one commit, or undo them all. */ PureResult pure_app_save(PureApp *); +/* Defer saving to pure_app_save (false: save after each event). Needs a data + * host that saves documents together (PureData.save). */ +bool pure_app_defer(PureApp *, bool deferred); /* Leaves unsaved frame changes unsaved: call pure_app_save first to keep them. */ void pure_app_stop(PureApp *); PureResult pure_app_frame(PureApp *); @@ -176,9 +193,11 @@ PureResult pure_app_input(PureApp *, uint32_t input, PureEdge); PureResult pure_app_text(PureApp *, uint32_t input, PureSpan text); /* Deliver a message. base is the revision the sender edited, or -1 for none; * a handler that edits the document needs one. payload stays alive until - * pure_app_finish. On return app->revision is the document's revision. */ + * pure_app_finish. On return app->revision is the document's revision + * (deferred: the revision the next save gives it). */ PureResult pure_app_message(PureApp *, uint32_t input, const char *key, int64_t base, PureSpan payload); -/* Visit the reply of the last message in order, without copying. */ +/* Visit the reply of the last message in order, without copying, until the + * next event (deferred: copy it before a later save too). */ PureResult pure_app_reply(const PureApp *, bool (*visit)(void *, PureSpan), void *context); void pure_app_finish(PureApp *); /* A number view; false for a text view or a failed evaluation. */ diff --git a/prototypes/connections/pure-run.c b/prototypes/connections/pure-run.c index 7b6e253..6e4a6e1 100644 --- a/prototypes/connections/pure-run.c +++ b/prototypes/connections/pure-run.c @@ -8,7 +8,8 @@ /* Test host for a verified module: reads commands from stdin and prints one * JSON snapshot per command. Not a product interface. With --store the app's * states live in that tree store: it starts from them, and frame changes are - * saved at the end of input or by save. + * saved at the end of input or by save. --batch N defers saving and saves + * every N commands (and at the end), so N events share one commit. * down NAME | up NAME | frame [N] | type NAME HEX|- | replace PATH | save | show * Text states and text views print as JSON strings; cursors as byte positions; * lists as arrays of objects; indices as positions; a table view as an array @@ -134,9 +135,11 @@ static int hex(char c) { int main(int argc, char **argv) { static PureModule modules[2]; - bool stored = argc == 4 && !strcmp(argv[2], "--store"); - if ((argc != 2 && !stored) || !load(argv[1], &modules[0])) { - if (argc != 2 && !stored) fputs("pure-run MODULE [--store PATH] < commands\n", stderr); + bool stored = argc >= 4 && !strcmp(argv[2], "--store"); + long batch = argc == 6 && stored && !strcmp(argv[4], "--batch") ? strtol(argv[5], NULL, 10) : 0; + bool usage = argc != 2 && !(stored && argc == 4) && !(batch > 0 && batch <= 100000); + if (usage || !load(argv[1], &modules[0])) { + if (usage) fputs("pure-run MODULE [--store PATH [--batch N]] < commands\n", stderr); return 2; } if (stored && pure_store_init(argv[3]) != PURE_OK) { fprintf(stderr, "cannot open store %s\n", argv[3]); return 2; } @@ -144,6 +147,8 @@ int main(int argc, char **argv) { PureApp app; PureResult started = pure_app_start(&app, &modules[active], stored ? pure_tree_data(argv[3]) : (PureData){0}); if (started != PURE_OK) { fprintf(stderr, "cannot start (%d)\n", (int)started); return 2; } + if (batch && !pure_app_defer(&app, true)) { fputs("cannot defer saving\n", stderr); return 2; } + long unsaved = 0; static char line[1 << 16]; static unsigned char text[1 << 15]; int status = 0; @@ -176,6 +181,10 @@ int main(int argc, char **argv) { } else if (!strcmp(command, "save")) { if (pure_app_save(&app) != PURE_OK) failure = "save failed"; } else if (strcmp(command, "show")) failure = "unknown command"; + if (batch && strcmp(command, "show") && ++unsaved == batch) { + unsaved = 0; + if (pure_app_save(&app) != PURE_OK) failure = "save failed"; + } if (failure) status = 1; show(&app, failure); } diff --git a/prototypes/connections/test-module.c b/prototypes/connections/test-module.c index 64409b2..e1d1ccf 100644 --- a/prototypes/connections/test-module.c +++ b/prototypes/connections/test-module.c @@ -1574,6 +1574,144 @@ static void stored_states(void) { stored_migration(); } +/* ---- Deferred saving ---------------------------------------------------- */ + +typedef struct { uint32_t input; const char *word; } Step; + +static PureResult apply_step(PureApp *app, Step step) { + if (step.word) return pure_app_text(app, step.input, (PureSpan){(const unsigned char *)step.word, strlen(step.word)}); + return pure_app_input(app, step.input, PURE_EDGE_DOWN); +} + +/* Deferred events apply at once and last only through a save: an app that + * stops without saving comes back as it was at the last save. The oracle is + * an app in memory that replays exactly the saved events. */ +static void deferred_equivalence(Spec spec, const char *const *words, size_t word_count, uint32_t seed) { + char *path = new_store(); + CHECK(load(spec, &module_a) == PURE_OK); + static Step history[2000]; + size_t saved = 0, applied = 0; + PureApp memory, stored; + CHECK(pure_app_start(&memory, &module_a, (PureData){0}) == PURE_OK); + CHECK(pure_app_start(&stored, &module_a, pure_tree_data(path)) == PURE_OK); + CHECK(pure_app_defer(&stored, true) && !pure_app_defer(&memory, true)); + unsigned lost = 0, saves = 0; + for (unsigned step = 0; step < 600; ++step) { + Step next = {next_random(&seed) % module_a.inputs, NULL}; + if (module_a.input[next.input].kind == PURE_KIND_TEXT) next.word = words[next_random(&seed) % word_count]; + PureResult a = apply_step(&memory, next), b = apply_step(&stored, next); + CHECK(a == b); + if (a == PURE_OK) history[applied++] = next; + same_states(&memory, &stored); + uint32_t roll = next_random(&seed) % 16; + if (roll < 3) { + CHECK(pure_app_save(&stored) == PURE_OK); + saved = applied; + ++saves; + } else if (roll == 3) { + /* Stop without saving: only saved events come back. */ + pure_app_stop(&stored); + pure_app_stop(&memory); + CHECK(pure_app_start(&stored, &module_a, pure_tree_data(path)) == PURE_OK && pure_app_defer(&stored, true)); + CHECK(pure_app_start(&memory, &module_a, (PureData){0}) == PURE_OK); + for (size_t k = 0; k < saved; ++k) CHECK(apply_step(&memory, history[k]) == PURE_OK); + lost += (unsigned)(applied - saved); + applied = saved; + same_states(&memory, &stored); + } + } + CHECK(saves > 50 && lost > 0); + pure_app_stop(&memory); + pure_app_stop(&stored); + remove_store(path); +} + +/* A burst of events is saved with one commit: when that commit fails every + * event of the burst is undone, and a process that dies while saving comes + * back with all of the burst or none of it. */ +static void deferred_bursts(void) { + char *path = new_store(); + Spec spec = {crash_nums, 1, crash_inputs, 1, NULL, 0, crash_handlers, 1, crash_code, LEN(crash_code), NULL, 0, + crash_states, 4, crash_constants, 2, NULL, 0, crash_records, 1, NULL, 0}; + CHECK(load(spec, &module_a) == PURE_OK); + PureApp app; + CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pure_app_defer(&app, true)); + for (int i = 0; i < 5; ++i) CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK); + CHECK(consistent(&app) == 5 && app.events == 5); + for (fail_stage = PURE_AFTER_HEADER; fail_stage <= PURE_BEFORE_RENAME; fail_stage += 2) { + CHECK(pure_app_save(&app) == PURE_STORE_ERROR && consistent(&app) == 0 && app.events == 0); + for (int i = 0; i < 5; ++i) CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK); + } + fail_stage = -1; + CHECK(pure_app_save(&app) == PURE_OK && consistent(&app) == 5); + pure_app_stop(&app); + CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && consistent(&app) == 5); + pure_app_stop(&app); + unsigned expected = 5; + for (int stage = PURE_AFTER_HEADER; stage <= PURE_BEFORE_DIR_SYNC; ++stage) { + pid_t child = fork(); + CHECK(child >= 0); + if (!child) { + PureApp doomed; + if (pure_app_start(&doomed, &module_a, pure_tree_data(path)) != PURE_OK || !pure_app_defer(&doomed, true)) _exit(1); + for (int i = 0; i < 7; ++i) if (pure_app_input(&doomed, 0, PURE_EDGE_DOWN) != PURE_OK) _exit(2); + crash = 1; + fail_stage = stage; + pure_app_save(&doomed); + _exit(0); + } + int status; + CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 77); + if (stage == PURE_BEFORE_DIR_SYNC) expected += 7; + CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && consistent(&app) == expected); + pure_app_stop(&app); + } + remove_store(path); +} + +/* Messages in a burst: documents of different keys commit together with the + * states, each moving one revision; a second message for a key saves the + * burst first, and so does a full burst. */ +static void deferred_messages(void) { + char *path = new_store(); + CHECK(load(editor_spec, &module_a) == PURE_OK); + PureApp app; + unsigned char buffer[64]; + char key[8]; + CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pure_app_defer(&app, true)); + for (int i = 0; i < 3; ++i) { + snprintf(key, sizeof(key), "k%d", i); + CHECK(pure_app_message(&app, 1, key, 0, edit_payload(buffer, 0, 0, "ab")) == PURE_OK && app.revision == 1); + } + CHECK(app.values[0] == 3 && app.pending_count == 3 && app.open); + /* The same key again: the burst is saved first, then this edit joins a new one. */ + CHECK(pure_app_message(&app, 1, "k0", 1, edit_payload(buffer, 2, 0, "c")) == PURE_OK && app.revision == 2); + CHECK(app.pending_count == 1 && app.values[0] == 4); + CHECK(pure_app_message(&app, 1, "k1", 0, edit_payload(buffer, 0, 0, "x")) == PURE_CONFLICT); + CHECK(pure_app_save(&app) == PURE_OK && !app.open && !app.pending_count); + for (int i = 0; i < PURE_APP_PENDING + 1; ++i) { + snprintf(key, sizeof(key), "m%d", i); + CHECK(pure_app_message(&app, 1, key, 0, edit_payload(buffer, 0, 0, "y")) == PURE_OK); + CHECK(app.pending_count == (i < PURE_APP_PENDING ? (unsigned)i + 1 : 1)); /* full: saved on the way */ + } + pure_app_stop(&app); /* the last edit was never saved */ + CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && app.values[0] == 4 + PURE_APP_PENDING); + CHECK(pure_app_message(&app, 0, "k0", -1, (PureSpan){0}) == PURE_OK && app.revision == 2 && replied(&app, "abc")); + CHECK(pure_app_message(&app, 0, "m16", -1, (PureSpan){0}) == PURE_NOT_FOUND); + CHECK(pure_app_message(&app, 0, "m15", -1, (PureSpan){0}) == PURE_OK && replied(&app, "y")); + pure_app_stop(&app); + remove_store(path); +} + +static void deferred(void) { + static const char *const list_words[] = {"a", "\xea\xb0\x80", "", "\n", "\xf0\x9f\x98\x80"}; + static const char *const text_words[] = {"a", "\xea\xb0\x80", "\n", "\xf0\x9f\x98\x80", "xy\nz", ""}; + deferred_equivalence(list_spec, list_words, LEN(list_words), 0x2545f491u); + deferred_equivalence(text_spec, text_words, LEN(text_words), 0x1b873593u); + deferred_bursts(); + deferred_messages(); +} + int main(void) { behaviour(); faults(); @@ -1588,7 +1726,8 @@ int main(void) { tables(); table_mutations(); stored_states(); + deferred(); printf("{\"sizeof_module\":%zu,\"sizeof_app\":%zu}\n", sizeof(PureModule), sizeof(PureApp)); - puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration\"}"); + puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration-deferred-bursts\"}"); return 0; } |