yonote/yjs
24
0
Fork 0
Code Pull Requests1 Releases Activity

late join works (also when activating garbage collector), added some tests to verify (havent tested for large >500 operations)

Browse Source
This commit is contained in:
Kevin Jahns 2015-10-12 15:17:12 +02:00
parent 9c4074e3e3
commit e32aef4c9f
10 changed files with 178 additions and 83 deletions

2
package.json

@ -4,7 +4,7 @@
"description": "A framework for real-time p2p shared editing on arbitrary complex data types",
"main": "y.js",
"scripts": {
"test": "./node_modules/.bin/gulp test",
"test": "node --harmony ./node_modules/.bin/gulp test",
"lint": "./node_modules/.bin/standard",
"build": "./node_modules/.bin/standard build"
},

4
src/Connector.js

@ -156,13 +156,13 @@ class AbstractConnector {
var currentStateSet = yield* this.getStateSet()
yield* this.applyDeleteSet(m.deleteSet)
var ds = yield* this.getDeleteSet()
var ops = yield* this.getOperations(m.stateSet)
ops = JSON.parse(JSON.stringify(ops)) // TODO: don't do something like that!!
conn.send(sender, {
type: 'sync step 2',
os: ops,
stateSet: currentStateSet,
deleteSet: yield* this.getDeleteSet()
deleteSet: ds
})
if (this.forwardToSyncingClients) {
conn.syncingClients.push(sender)

20
src/Connectors/Test.js

@ -1,4 +1,4 @@
/* global getRandom, Y, wait */
/* global getRandom, Y, wait, async */
'use strict'
var globalRoom = {
@ -82,19 +82,25 @@ class Test extends Y.AbstractConnector {
globalRoom.addUser(this)
super.reconnect()
}
return this.flushAll()
}
disconnect () {
if (!this.isDisconnected()) {
globalRoom.removeUser(this.userId)
super.disconnect()
}
return wait()
}
flush () {
var buff = globalRoom.buffers[this.userId]
while (buff.length > 0) {
var m = buff.shift()
this.receiveMessage(m[0], m[1])
}
var self = this
return async(function * () {
yield wait()
while (globalRoom.buffers[self.userId].length > 0) {
var m = globalRoom.buffers[self.userId].shift()
this.receiveMessage(m[0], m[1])
yield wait()
}
})
}
flushAll () {
return new Promise(function (resolve) {
@ -115,7 +121,7 @@ class Test extends Y.AbstractConnector {
}
// in the case that there are
// still actions that want to be performed
wait(0).then(nextFlush)
wait().then(nextFlush)
})
}
/*

75
src/Helper.spec.js

@ -66,56 +66,68 @@ function getRandomNumber (n) {
}
g.getRandomNumber = getRandomNumber
g.applyRandomTransactions = async(function * applyRandomTransactions (users, objects, transactions, numberOfTransactions) {
function * applyTransactions (relAmount, numberOfTransactions, objects, users, transactions) {
function randomTransaction (root) {
var f = getRandom(transactions)
f(root)
}
function * applyTransactions (relAmount) {
for (var i = 0; i < numberOfTransactions * relAmount + 1; i++) {
var r = Math.random()
if (r >= 0.9) {
// 10% chance to flush
users[0].connector.flushOne() // flushes for some user.. (not necessarily 0)
} else if (r >= 0.1) {
// 80% chance to create operation
randomTransaction(getRandom(objects))
for (var i = 0; i < numberOfTransactions * relAmount + 1; i++) {
var r = Math.random()
if (r >= 0.5) {
// 50% chance to flush
users[0].connector.flushOne() // flushes for some user.. (not necessarily 0)
} else if (r >= 0.05) {
// 45% chance to create operation
randomTransaction(getRandom(objects))
} else {
// 5% chance to disconnect/reconnect
var u = getRandom(users)
if (u.connector.isDisconnected()) {
yield u.reconnect()
} else {
// 10% chance to disconnect/reconnect
var u = getRandom(users)
if (u.connector.isDisconnected()) {
u.reconnect()
} else {
u.disconnect()
}
yield u.disconnect()
}
yield wait()
}
yield wait()
}
yield* applyTransactions(0.5)
}
g.applyRandomTransactionsAllRejoinNoGC = async(function * applyRandomTransactions (users, objects, transactions, numberOfTransactions) {
yield* applyTransactions(1, numberOfTransactions, objects, users, transactions)
yield users[0].connector.flushAll()
yield g.garbageCollectAllUsers(users)
yield wait()
users[0].disconnect()
yield wait()
yield* applyTransactions(0.5)
yield users[0].connector.flushAll()
yield wait(50)
for (var u in users) {
users[u].reconnect()
yield users[u].reconnect()
}
yield wait(100)
yield users[0].connector.flushAll()
yield g.garbageCollectAllUsers(users)
})
g.applyRandomTransactionsWithGC = async(function * applyRandomTransactions (users, objects, transactions, numberOfTransactions) {
yield* applyTransactions(1, numberOfTransactions, objects, users.slice(1), transactions)
yield users[0].connector.flushAll()
yield g.garbageCollectAllUsers(users)
yield wait(100)
for (var u in users) {
// TODO: here, we enforce that two users never sync at the same time with u[0]
// enforce that in the connector itself!
yield users[u].reconnect()
}
yield wait(100)
yield users[0].connector.flushAll()
yield wait(100)
yield g.garbageCollectAllUsers(users)
})
g.garbageCollectAllUsers = async(function * garbageCollectAllUsers (users) {
return yield wait(100)// TODO!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
/*
// gc two times because of the two gc phases (really collect everything)
yield wait(100)
for (var i in users) {
yield users[i].db.garbageCollect()
yield users[i].db.garbageCollect()
}
*/
yield wait(100)
})
g.compareAllUsers = async(function * compareAllUsers (users) {
@ -142,7 +154,6 @@ g.compareAllUsers = async(function * compareAllUsers (users) {
})
}
yield users[0].connector.flushAll()
// gc two times because of the two gc phases (really collect everything)
yield g.garbageCollectAllUsers(users)
for (var uid = 0; uid < users.length; uid++) {
@ -178,6 +189,8 @@ g.compareAllUsers = async(function * compareAllUsers (users) {
u.db.requestTransaction(t1)
yield wait()
u.db.os.iterate(null, null, function (o) {
o = Y.utils.copyObject(o)
delete o.origin
db1.push(o)
})
} else {
@ -188,6 +201,8 @@ g.compareAllUsers = async(function * compareAllUsers (users) {
expect(ds1).toEqual(ds2) // exported structure
var count = 0
u.db.os.iterate(null, null, function (o) {
o = Y.utils.copyObject(o)
delete o.origin
expect(db1[count++]).toEqual(o)
})
}

61
src/OperationStore.js

@ -158,6 +158,15 @@ class AbstractTransaction {
}
}
}
/*
Really remove an op and all its effects.
The complicated case here is the Insert operation:
* reset left
* reset right
* reset parent.start
* reset parent.end
* reset origins of all right ops
*/
* garbageCollectOperation (id) {
// check to increase the state of the respective user
var state = yield* this.getState(id[0])
@ -185,11 +194,45 @@ class AbstractTransaction {
yield* this.setOperation(left)
}
// remove gc'd op from the right op, if it exists
// also reset origins of right ops
if (o.right != null) {
var right = yield* this.getOperation(o.right)
right.left = o.left
if (Y.utils.compareIds(right.origin, o.id)) { // rights origin is o
// find new origin of right ops
// origin is the first left deleted operation
var neworigin = o.left
while (neworigin != null) {
var neworigin_ = yield* this.getOperation(neworigin)
if (neworigin_.deleted) {
break
}
neworigin = neworigin_.left
}
// reset origin of right
right.origin = neworigin
// reset origin of all right ops (except first right - duh!),
// until you find origin pointer to the left of o
var i = right.right == null ? null : yield* this.getOperation(right.right)
var ids = [o.id, o.right]
while (i != null && ids.some(function (id) {
return Y.utils.compareIds(id, i.origin)
})) {
if (Y.utils.compareIds(i.origin, o.id)) {
// reset origin of i
i.origin = neworigin
yield* this.setOperation(i)
}
// get next i
i = i.right == null ? null : yield* this.getOperation(i.right)
}
} /* otherwise, rights origin is to the left of o,
then there is no right op (from o), that origins in o */
yield* this.setOperation(right)
}
// remove gc'd op from parent, if it exists
var parent = yield* this.getOperation(o.parent)
var setParent = false // whether to save parent to the os
@ -255,14 +298,16 @@ class AbstractOperationStore {
function garbageCollect () {
return new Promise((resolve) => {
os.requestTransaction(function * () {
for (var i in os.gc2) {
var oid = os.gc2[i]
yield* this.garbageCollectOperation(oid)
}
os.gc2 = os.gc1
os.gc1 = []
if (os.gcTimeout > 0) {
os.gcInterval = setTimeout(garbageCollect, os.gcTimeout)
if (os.y.connector.isSynced) {
for (var i in os.gc2) {
var oid = os.gc2[i]
yield* this.garbageCollectOperation(oid)
}
os.gc2 = os.gc1
os.gc1 = []
if (os.gcTimeout > 0) {
os.gcInterval = setTimeout(garbageCollect, os.gcTimeout)
}
}
resolve()
})

13
src/OperationStores/Memory.js

@ -4,6 +4,7 @@
class DeleteStore extends Y.utils.RBTree {
constructor () {
super()
this.mem = []
}
isDeleted (id) {
var n = this.findNodeWithUpperBound(id)
@ -15,6 +16,7 @@ class DeleteStore extends Y.utils.RBTree {
returns the delete node
*/
markGarbageCollected (id) {
// this.mem.push(["gc", id]);
var n = this.markDeleted(id)
if (!n.val.gc) {
if (n.val.id[1] < id[1]) {
@ -23,6 +25,9 @@ class DeleteStore extends Y.utils.RBTree {
n.val.len -= newlen
n = this.add({id: id, len: newlen, gc: false})
}
// get prev&next before adding a new operation
var prev = n.prev()
var next = n.next()
if (id[1] < n.val.id[1] + n.val.len - 1) {
// un-extend right
this.add({id: [id[0], id[1] + 1], len: n.val.len - 1, gc: false})
@ -30,8 +35,6 @@ class DeleteStore extends Y.utils.RBTree {
}
// set gc'd
n.val.gc = true
var prev = n.prev()
var next = n.next()
// can extend left?
if (
prev != null &&
@ -52,7 +55,6 @@ class DeleteStore extends Y.utils.RBTree {
super.delete(next.val.id)
}
}
return n
}
/*
Mark an operation as deleted.
@ -60,6 +62,7 @@ class DeleteStore extends Y.utils.RBTree {
returns the delete node
*/
markDeleted (id) {
// this.mem.push(["del", id]);
var n = this.findNodeWithUpperBound(id)
if (n != null && n.val.id[0] === id[0]) {
if (n.val.id[1] <= id[1] && id[1] < n.val.id[1] + n.val.len) {
@ -85,8 +88,10 @@ class DeleteStore extends Y.utils.RBTree {
) {
n.val.len = n.val.len + next.val.len
super.delete(next.val.id)
return this.findNode(n.val.id)
} else {
return n
}
return n
}
/*
A DeleteSet (ds) describes all the deleted ops in the OS

30
src/OperationStores/Memory.spec.js

@ -83,10 +83,8 @@ describe('Memory', function () {
})
it('Debug #5', async(function * (done) {
var store = new Y.Memory(null, {
db: {
name: 'Memory',
gcTimeout: -1
}
name: 'Memory',
gcTimeout: -1
})
store.requestTransaction(function * () {
yield* this.applyDeleteSet({'16': [[1, 2, false]], '17': [[0, 1, true], [1, 3, false]]})
@ -98,10 +96,8 @@ describe('Memory', function () {
}))
it('Debug #6', async(function * (done) {
var store = new Y.Memory(null, {
db: {
name: 'Memory',
gcTimeout: -1
}
name: 'Memory',
gcTimeout: -1
})
store.requestTransaction(function * () {
yield* this.applyDeleteSet({'40': [[0, 3, false]]})
@ -111,5 +107,23 @@ describe('Memory', function () {
done()
})
}))
it('Debug #7', function () {
ds.markDeleted(['9', 2])
ds.markDeleted(['11', 2])
ds.markDeleted(['11', 4])
ds.markDeleted(['11', 1])
ds.markDeleted(['9', 4])
ds.markDeleted(['10', 0])
ds.markGarbageCollected(['11', 2])
ds.markDeleted(['11', 2])
ds.markGarbageCollected(['11', 3])
ds.markDeleted(['11', 3])
ds.markDeleted(['11', 3])
ds.markDeleted(['9', 4])
ds.markDeleted(['10', 0])
ds.markGarbageCollected(['11', 1])
ds.markDeleted(['11', 1])
expect(ds.toDeleteSet()).toEqual({'9': [[2, 1, false], [4, 1, false]], '10': [[0, 1, false]], '11': [[1, 3, true], [4, 1, false]]})
})
})
})

38
src/Types/Array.spec.js

@ -1,7 +1,7 @@
/* global createUsers, wait, Y, compareAllUsers, getRandomNumber, applyRandomTransactions, async, garbageCollectAllUsers, describeManyTimes */
/* global createUsers, wait, Y, compareAllUsers, getRandomNumber, applyRandomTransactionsAllRejoinNoGC, applyRandomTransactionsWithGC, async, garbageCollectAllUsers, describeManyTimes */
/* eslint-env browser,jasmine */
var numberOfYArrayTests = 10
var numberOfYArrayTests = 100
var repeatArrayTests = 1
describe('Array Type', function () {
@ -71,11 +71,8 @@ describe('Array Type', function () {
l2 = yield y2.get('Array')
l2.insert(1, [2])
l2.insert(1, [3])
yield flushAll()
yconfig2.reconnect()
yconfig3.reconnect()
yield wait()
yield flushAll()
yield yconfig2.reconnect()
yield yconfig3.reconnect()
expect(l1.toArray()).toEqual(l2.toArray())
done()
}))
@ -84,15 +81,12 @@ describe('Array Type', function () {
l1 = yield y1.set('Array', Y.Array)
l1.insert(0, ['x', 'y'])
yield flushAll()
yconfig2.disconnect()
yield yconfig2.disconnect()
yield wait()
l2 = yield y2.get('Array')
l2.delete(1, 1)
l1.delete(0, 2)
yield flushAll()
yconfig2.reconnect()
yield wait()
yield flushAll()
yield yconfig2.reconnect()
expect(l1.toArray()).toEqual(l2.toArray())
done()
}))
@ -130,7 +124,7 @@ describe('Array Type', function () {
l1.delete(0, 3)
l2 = yield y2.get('Array')
yield wait()
yconfig2.reconnect()
yield yconfig2.reconnect()
yield wait()
l3 = yield y3.get('Array')
yield flushAll()
@ -148,7 +142,7 @@ describe('Array Type', function () {
yconfig1.disconnect()
l1.delete(0, 3)
l2 = yield y2.get('Array')
yconfig1.reconnect()
yield yconfig1.reconnect()
l3 = yield y3.get('Array')
yield flushAll()
expect(l1.toArray()).toEqual(l2.toArray())
@ -188,7 +182,7 @@ describe('Array Type', function () {
l1.delete(0, 3)
l2 = yield y2.get('Array')
yield wait()
yconfig1.reconnect()
yield yconfig1.reconnect()
yield wait()
l3 = yield y3.get('Array')
yield flushAll()
@ -238,11 +232,21 @@ describe('Array Type', function () {
expect(this.arrays.length).toEqual(this.users.length)
done()
}))
it(`succeed after ${numberOfYArrayTests} actions`, async(function * (done) {
it(`succeed after ${numberOfYArrayTests} actions, no GC, all users disconnecting/reconnecting`, async(function * (done) {
for (var u of this.users) {
u.connector.debug = true
}
yield applyRandomTransactions(this.users, this.arrays, randomArrayTransactions, numberOfYArrayTests)
yield applyRandomTransactionsAllRejoinNoGC(this.users, this.arrays, randomArrayTransactions, numberOfYArrayTests)
yield flushAll()
yield compareArrayValues(this.arrays)
yield compareAllUsers(this.users)
done()
}))
it(`succeed after ${numberOfYArrayTests} actions, GC, user[0] is not disconnecting`, async(function * (done) {
for (var u of this.users) {
u.connector.debug = true
}
yield applyRandomTransactionsWithGC(this.users, this.arrays, randomArrayTransactions, numberOfYArrayTests)
yield flushAll()
yield compareArrayValues(this.arrays)
yield compareAllUsers(this.users)

12
src/Types/Map.spec.js

@ -1,4 +1,4 @@
/* global createUsers, Y, compareAllUsers, getRandomNumber, applyRandomTransactions, async, describeManyTimes */
/* global createUsers, Y, compareAllUsers, getRandomNumber, applyRandomTransactionsAllRejoinNoGC, applyRandomTransactionsWithGC, async, describeManyTimes */
/* eslint-env browser,jasmine */
var numberOfYMapTests = 100
@ -201,8 +201,14 @@ describe('Map Type', function () {
this.maps = yield Promise.all(promises)
done()
}))
it(`succeed after ${numberOfYMapTests} actions`, async(function * (done) {
yield applyRandomTransactions(this.users, this.maps, randomMapTransactions, numberOfYMapTests)
it(`succeed after ${numberOfYMapTests} actions, no GC, all users disconnecting/reconnecting`, async(function * (done) {
yield applyRandomTransactionsAllRejoinNoGC(this.users, this.maps, randomMapTransactions, numberOfYMapTests)
yield flushAll()
yield compareMapValues(this.maps)
done()
}))
it(`succeed after ${numberOfYMapTests} actions, GC, user[0] is not disconnecting`, async(function * (done) {
yield applyRandomTransactionsWithGC(this.users, this.maps, randomMapTransactions, numberOfYMapTests)
yield flushAll()
yield compareMapValues(this.maps)
done()

6
src/y.js

@ -33,13 +33,13 @@ class YConfig {
return this.connector.isSynced
}
disconnect () {
this.connector.disconnect()
return this.connector.disconnect()
}
reconnect () {
this.connector.reconnect()
return this.connector.reconnect()
}
destroy () {
this.connector.disconnect()
this.disconnect()
this.db.destroy()
this.connector = null
this.db = null