Merge pull request #1 from y-js/master

Updating to latest yjs version
This commit is contained in:
István Koren 2016-02-13 13:14:43 +01:00
commit fa9ff669e4
13 changed files with 791 additions and 351 deletions

184
README.md
View File

@ -1,49 +1,55 @@
# ![Yjs](http://y-js.org/images/yjs.png) # ![Yjs](http://y-js.org/images/yjs.png)
Yjs is a framework for optimistic concurrency control and automatic conflict resolution on shared data types. The framework implements a new OT-like concurrency algorithm and provides similar functionality as [ShareJs] and [OpenCoweb]. Yjs was designed to handle concurrent actions on arbitrary complex data types like Text, Json, and XML. We provide a tutorial and some applications for this framework on our [homepage](http://y-js.org/). Yjs is a framework for optimistic concurrency control and automatic conflict resolution on shared data. The framework provides similar functionality as [ShareJs] and [OpenCoweb], but supports peer-to-peer communication protocols by default. Yjs was designed to handle concurrent actions on arbitrary data like Text, Json, and XML. We also provide support for storing and manipulating your shared data offline. For more information and demo applications visit our [homepage](http://y-js.org/).
**NOTE** This project is currently migrating. So there may exist some information that is not true anymore.. You can create you own shared types easily.
Therefore, you can design the sturcture of your custom type,
You can create you own shared types easily. Therefore, you can take matters into your own hand by defining the meaning of the shared types and ensure that it is valid, while Yjs ensures data consistency (everyone will eventually end up with the same data). We already provide data types for and ensure data validity, while Yjs ensures data consistency (everyone will eventually end up with the same data).
We already provide abstract data types for
| Name | Description | | Name | Description |
|----------|-------------------| |----------|-------------------|
| map | Add, update, and remove properties of an object. Included in Yjs| |[map](https://github.com/y-js/y-map) | A shared Map implementation. Maps from text to any stringify-able object |
|[array](https://github.com/y-js/y-array) | A shared linked list implementation | |[array](https://github.com/y-js/y-array) | A shared Array implementation |
|[selections](https://github.com/y-js/y-selections) | Manages selections on types that use linear structures (e.g. the y-array type). Select a range of elements, and assign meaning to them.| |[xml](https://github.com/y-js/y-xml) | An implementation of the DOM. You can create a two way binding to Browser DOM objects |
|[xml](https://github.com/y-js/y-xml) | An implementation of the DOM. You can create a two way binding to Browser DOM objects| |[text](https://github.com/y-js/y-text) | Collaborate on text. Supports two way binding to textareas, input elements, or HTML elements (e.g. <*h1*>, or <*p*>). Also supports the [Ace Editor](https://ace.c9.io) |
|[text](https://github.com/y-js/y-text) | Collaborate on text. Supports two way binding to textareas, input elements, or HTML elements (e.g. *h1*, or *p*)| |[richtext](https://github.com/y-js/y-richtext) | Collaborate on rich text. Supports two way binding to the [Quill Rich Text Editor](http://quilljs.com/)|
|[richtext](https://github.com/y-js/y-richtext) | Collaborate on rich text. Supports two way binding to several editors|
Unlike other frameworks, Yjs supports P2P message propagation and is not bound to a specific communication protocol. Therefore, Yjs is extremely scalable and can be used in a wide range of application scenarios. Yjs supports P2P message propagation, and is not bound to a specific communication protocol. Therefore, Yjs is extremely scalable and can be used in a wide range of application scenarios.
We support several communication protocols as so called *Connectors*. You can create your own connector too - read [this wiki page](https://github.com/y-js/yjs/wiki/Custom-Connectors). Currently, we support the following communication protocols: We support several communication protocols as so called *Connectors*.
You can create your own connector too - read [this wiki page](https://github.com/y-js/yjs/wiki/Custom-Connectors).
Currently, we support the following communication protocols:
|Name | Description | |Name | Description |
|----------------|-----------------------------------| |----------------|-----------------------------------|
|[xmpp](https://github.com/y-js/y-xmpp) | Propagate updates in a XMPP multi-user-chat room ([XEP-0045](http://xmpp.org/extensions/xep-0045.html))| |[xmpp](https://github.com/y-js/y-xmpp) | Propagate updates in a XMPP multi-user-chat room ([XEP-0045](http://xmpp.org/extensions/xep-0045.html))|
|[webrtc](https://github.com/y-js/y-webrtc) | Propagate updates Browser2Browser via WebRTC| |[webrtc](https://github.com/y-js/y-webrtc) | Propagate updates Browser2Browser via WebRTC|
|[websockets](https://github.com/y-js/y-websockets-client) | Exchange updates efficiently in the classical client-server model |
|[test](https://github.com/y-js/y-test) | A Connector for testing purposes. It is designed to simulate delays that happen in worst case scenarios| |[test](https://github.com/y-js/y-test) | A Connector for testing purposes. It is designed to simulate delays that happen in worst case scenarios|
You are not limited to use a specific database to store the shared data. We provide the following database adapters:
|Name | Description |
|----------------|-----------------------------------|
|[memory](https://github.com/y-js/y-memory) | In-memory storage. |
|[indexeddb](https://github.com/y-js/y-indexeddb) | Offline storage for the browser |
You can use Yjs client-, and server- side. You can get it as via npm, and bower. We even provide polymer elements for Yjs! You can use Yjs client-, and server- side. You can get it as via npm, and bower. We even provide polymer elements for Yjs!
The advantages over similar frameworks are support for The advantages over similar frameworks are support for
* .. P2P message propagation and arbitrary communication protocols * .. P2P message propagation and arbitrary communication protocols
* .. arbitrary complex data types * .. share any type of data. The types provide a convenient interface
* .. offline editing: Changes are stored persistently and only relevant changes are propagated on rejoin * .. offline support: Changes are stored persistently and only relevant changes are propagated on rejoin
* .. AnyUndo: Undo *any* action that was executed in constant time (coming..)
* .. Intention Preservation: When working on Text, the intention of your changes are preserved. This is particularily important when working offline. Every type has a notion on how we define Intention Preservation on it. * .. Intention Preservation: When working on Text, the intention of your changes are preserved. This is particularily important when working offline. Every type has a notion on how we define Intention Preservation on it.
## Use it! ## Use it!
You can find a tutorial, and examples on the [website](http://y-js.org). Furthermore, the [github wiki](https://github.com/y-js/yjs/wiki) offers more information about how you can use Yjs in your application. Install yjs and its modules with [bower](http://bower.io/), or with [npm](https://www.npmjs.org/package/yjs).
Either clone this git repository, install it with [bower](http://bower.io/), or install it with [npm](https://www.npmjs.org/package/yjs).
### Bower ### Bower
``` ```
bower install y-js/yjs bower install yjs
``` ```
Then you include the libraries directly from the installation folder. Then you include the libraries directly from the installation folder.
``` ```
@ -60,57 +66,108 @@ And use it like this with *npm*:
Y = require("yjs"); Y = require("yjs");
``` ```
# Y() # Text editing example
In order to create an instance of Y, you need to have a connection object (instance of a Connector). Then, you can create a shared data type like this:
``` ```
var y = new Y(connector); Y({
db: {
name: 'memory' // store in memory.
// name: 'indexeddb'
},
connector: {
name: 'websockets-client', // choose the websockets connector
// name: 'webrtc'
// name: 'xmpp'
room: 'Textarea-example-dev'
},
sourceDir: '/bower_components', // location of the y-* modules
share: {
textarea: 'Text' // y.share.textarea is of type Y.Text
}
// types: ['Richtext', 'Array'] // optional list of types you want to import
}).then(function (y) {
// bind the textarea to a shared text element
y.share.textarea.bind(document.getElementById('textfield'))
}
``` ```
# Api
# Y.Map ### Y(options)
Yjs includes only one type by default - the Y.Map type. It mimics the behaviour of a javascript Object. You can create, update, and remove properies on the Y.Map type. Furthermore, you can observe changes on this type as you can observe changes on Javascript Objects with [Object.observe](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Object/observe) - an ECMAScript 7 proposal which is likely to become accepted by the committee. Until then, we have our own implementation. * options.db
* Will be forwarded to the database adapter. Specify the database adaper on `options.db.name`.
* Have a look at the used database adapter repository to see all available options.
##### Reference * options.connector
* Create * Will be forwarded to the connector adapter. Specify the connector adaper on `options.connector.name`.
* All our connectors implement a `room` property. Clients that specify the same room share the same data.
* All of our connectors specify an `url` property that defines the connection endpoint of the used connector.
* All of our connectors also have a default connection endpoint that you can use for development.
* Have a look at the used connector repository to see all available options.
* options.sourceDir
* Path where all y-* modules are stored.
* Defaults to `/bower_components`
* Not required when running on `nodejs` / `iojs`
* When using browserify you can specify all used modules like this:
``` ```
var map = y.set("new_map", Y.Map).then(function(map){ var Y = require('yjs')
map // is my map type // you need to require the db, connector, and *all* types you use!
}); require('y-memory')(Y)
``` require('y-webrtc')(Y)
* Every instance of Y is an Y.Map require('y-map')(Y)
``` // ..
var y = new Y(options); ```
``` * options.share
* .get(name) * Specify on `options.share[arbitraryName]` types that are shared among all users.
* Retrieve the value of a property. If the value is a type, `.get(name)` returns a promise * E.g. Specify `options.share[arbitraryName] = 'Array'` to require y-array and create an Y.Array type on `y.share[arbitraryName]`.
* .set(name, value) * If userA doesn't specify `options.share[arbitraryName]`, it won't be available for userA.
* Set/update a property. `value` may be a primitive type, or a custom type definition (e.g. `Y.Map`) * If userB specifies `options.share[arbitraryName]`, it still won't be available for userA. But all the updates are send from userB to userA.
* .delete(name) * In contrast to Y.Map, types on `y.share.*` cannot be overwritten or deleted. Instead, they are merged among all users. This feature is only available on `y.share.*`
* Delete a property * Weird behavior: It is supported that two users specify different types with the same property name.
* .observe(observer) E.g. userA specifies `options.share.x = 'Array'`, and userB specifies `options.share.x = 'Text'`. But they'll only share data if they specified the same type with the same property name
* The `observer` is called whenever something on this object changes. Throws *add*, *update*, and *delete* events * options.type
* .observePath(path, observer) * Array of modules that Yjs needs to require, before instantiating a shared type.
* `path` is an array of property names. `observer` is called when the property under `path` is set, deleted, or updated * By default Yjs requires the specified database adapter, the specified connector, and all modules that are used in `options.share.*`
* .unobserve(f) * Put all types here that you intend to use, but are not used in y.share.*
* Delete an observer
# A note on intention preservation ### Instantiated Y object (y)
When users create/update/delete the same property concurrently, only one change will prevail. Changes on different properties do not conflict with each other. `Y(options)` returns a promise that is fulfilled when..
* All modules are loaded
* The specified database adapter is loaded
* The specified connector is loaded
* All types are included
* The connector is initialized, and a unique user id is set (received from the server)
* Note: When using y-indexeddb, a retrieved user id is stored on `localStorage`
The promise returns an instance of Y. We denote it with a lower case `y`.
* y.share.*
* Instances of the types you specified on options.share.*
* y.share.* can only be defined once when you instantiate Y!
* y.connector is an instance of Y.AbstractConnector
* y.connector.onUserEvent(function (event) {..})
* Observe user events (event.action is either 'userLeft' or 'userJoined')
* y.connector.whenSynced(listener)
* `listener` is executed when y synced with at least one user.
* `listener` is not called when no other user is in the same room.
* y-websockets-client aways waits to sync with the server
* y.connector.disconnect()
* Force to disconnect this instance from the other instances
* y.connector.reconnect()
* Try to reconnect to the other instances (needs to be supported by the connector)
* Not supported by y-xmpp
* y.destroy()
* Destroy this object.
* Destroys all types (they will throw weird errors if you still use them)
* Disconnects from the other instances (via connector)
* Removes all data from the database
* y.db.stopGarbageCollector()
* Stop the garbage collector. Call y.db.garbageCollect() to continue garbage collection
* y.db.gcTimeout :: Number (defaults to 50000 ms)
* Time interval between two garbage collect cycles
* It is required that all instances exchanged all messages after two garbage collect cycles (after 100000 ms per default)
* y.db.userId :: String
* The used user id for this client. **Never overwrite this**
# A note on time complexities
* .get(name)
* O(1)
* .set(name, value)
* O(1)
* .delete(name)
* O(1)
* Apply a delete operation from another user
* O(1)
* Apply an update operation from another user (set/update a property)
* Yjs does not transform against operations that do not conflict with each other.
* An operation conflicts with another operation if it changes the same property.
* Overall worst case complexety: O(|conflicts|!)
# Status # Status
Yjs is a work in progress. Different versions of the *y-* repositories may not work together. Just drop me a line if you run into troubles. Yjs is a work in progress. Different versions of the *y-* repositories may not work together. Just drop me a line if you run into troubles.
@ -144,3 +201,4 @@ Yjs is licensed under the [MIT License](./LICENSE.txt).
[ShareJs]: https://github.com/share/ShareJS [ShareJs]: https://github.com/share/ShareJS
[OpenCoweb]: https://github.com/opencoweb/coweb/wiki [OpenCoweb]: https://github.com/opencoweb/coweb/wiki

2
dist

@ -1 +1 @@
Subproject commit 96f8f77dc4d90cd7fdc09ecb354b8ce7ae270796 Subproject commit ef6d63c19af25d7d1f09f83b0487f0edf5bfe196

View File

@ -94,12 +94,12 @@ module.exports = function (gulp, helperOptions) {
return browserify({ return browserify({
entries: files.specs, entries: files.specs,
debug: options.debug debug: true
}).bundle() }).bundle()
.pipe(source('specs.js')) .pipe(source('specs.js'))
.pipe(buffer()) .pipe(buffer())
.pipe($.sourcemaps.init({loadMaps: true})) // .pipe($.sourcemaps.init({loadMaps: true}))
.pipe($.sourcemaps.write('.')) // .pipe($.sourcemaps.write('.'))
.pipe(gulp.dest('./build/')) .pipe(gulp.dest('./build/'))
}) })
@ -121,13 +121,14 @@ module.exports = function (gulp, helperOptions) {
gulp.task('updateSubmodule', function () { gulp.task('updateSubmodule', function () {
return gulp.src('./package.json', {read: false}) return gulp.src('./package.json', {read: false})
.pipe($.shell([ .pipe($.shell([
'git submodule update --init' 'git submodule update --init',
'cd dist && git pull origin dist'
])) ]))
}) })
gulp.task('bump', function () { gulp.task('bump', function () {
var bumptype var bumptype
return gulp.src(['./package.json', './dist/package.json', './dist/bower.json'], {base: '.'}) return gulp.src(['./package.json', './bower.json', './dist/bower.json'], {base: '.'})
.pipe($.prompt.prompt({ .pipe($.prompt.prompt({
type: 'checkbox', type: 'checkbox',
name: 'bump', name: 'bump',
@ -145,14 +146,14 @@ module.exports = function (gulp, helperOptions) {
gulp.task('publish', function (cb) { gulp.task('publish', function (cb) {
/* TODO: include 'test',*/ /* TODO: include 'test',*/
runSequence(['updateSubmodule', 'dist'], 'bump', function () { runSequence('updateSubmodule', 'dist', 'bump', function () {
return gulp.src('./package.json', {read: false}) return gulp.src('./package.json', {read: false})
.pipe($.prompt.confirm({ .pipe($.prompt.confirm({
message: 'Are you sure you want to publish this release?', message: 'Are you sure you want to publish this release?',
default: false default: false
})) }))
.pipe($.shell([ .pipe($.shell([
'cp ./README.md ./dist/', // 'cp README.md dist',
'standard', 'standard',
'echo "Deploying version <%= getVersion(file.path) %>"', 'echo "Deploying version <%= getVersion(file.path) %>"',
'git pull', 'git pull',
@ -163,6 +164,7 @@ module.exports = function (gulp, helperOptions) {
'cd ./dist/ && git push origin --tags', 'cd ./dist/ && git push origin --tags',
'git commit -am "Release <%= getVersion(file.path) %>" -n', 'git commit -am "Release <%= getVersion(file.path) %>" -n',
'git push', 'git push',
'npm publish',
'echo Finished <%= callback() %>' 'echo Finished <%= callback() %>'
], { ], {
templateData: { templateData: {

View File

@ -54,8 +54,9 @@ require('./gulpfile.helper.js')(gulp, {
moduleName: 'yjs', moduleName: 'yjs',
includeRuntime: true, includeRuntime: true,
specs: [ specs: [
'./src/Types/Map.spec.js', './src/Database.spec.js',
'./src/Database.spec.js' '../y-array/src/Array.spec.js',
'../y-map/src/Map.spec.js'
] ]
}) })
@ -65,7 +66,7 @@ gulp.task('dev:examples', ['watch:dist'], function () {
gulp.src(distfiles) gulp.src(distfiles)
.pipe($.watch(distfiles)) .pipe($.watch(distfiles))
.pipe($.rename(function (path) { .pipe($.rename(function (path) {
var dir = path.dirname.split('/')[0] var dir = path.dirname.split(/[\\\/]/)[0]
console.log(JSON.stringify(path)) console.log(JSON.stringify(path))
path.dirname = dir === '.' ? 'yjs' : dir path.dirname = dir === '.' ? 'yjs' : dir
})) }))

View File

@ -1,8 +1,8 @@
{ {
"name": "yjs", "name": "yjs",
"version": "0.6.42", "version": "0.8.18",
"description": "A framework for real-time p2p shared editing on arbitrary complex data types", "description": "A framework for real-time p2p shared editing on arbitrary complex data types",
"main": "y.js", "main": "./src/y.js",
"scripts": { "scripts": {
"test": "node --harmony ./node_modules/.bin/gulp test", "test": "node --harmony ./node_modules/.bin/gulp test",
"lint": "./node_modules/.bin/standard" "lint": "./node_modules/.bin/standard"
@ -71,6 +71,8 @@
"run-sequence": "^1.1.4", "run-sequence": "^1.1.4",
"standard": "^5.2.2", "standard": "^5.2.2",
"vinyl-buffer": "^1.0.0", "vinyl-buffer": "^1.0.0",
"vinyl-source-stream": "^1.1.0" "vinyl-source-stream": "^1.1.0",
} "babel-eslint": "^5.0.0-beta6"
},
"dependencies": {}
} }

View File

@ -50,6 +50,7 @@ module.exports = function (Y/* :any */) {
this.debug = opts.debug === true this.debug = opts.debug === true
this.broadcastedHB = false this.broadcastedHB = false
this.syncStep2 = Promise.resolve() this.syncStep2 = Promise.resolve()
this.broadcastOpBuffer = []
} }
reconnect () { reconnect () {
} }
@ -63,26 +64,32 @@ module.exports = function (Y/* :any */) {
return this.y.db.stopGarbageCollector() return this.y.db.stopGarbageCollector()
} }
setUserId (userId) { setUserId (userId) {
this.userId = userId if (this.userId == null) {
return this.y.db.setUserId(userId) this.userId = userId
return this.y.db.setUserId(userId)
} else {
return null
}
} }
onUserEvent (f) { onUserEvent (f) {
this.userEventListeners.push(f) this.userEventListeners.push(f)
} }
userLeft (user) { userLeft (user) {
delete this.connections[user] if (this.connections[user] != null) {
if (user === this.currentSyncTarget) { delete this.connections[user]
this.currentSyncTarget = null if (user === this.currentSyncTarget) {
this.findNextSyncTarget() this.currentSyncTarget = null
} this.findNextSyncTarget()
this.syncingClients = this.syncingClients.filter(function (cli) { }
return cli !== user this.syncingClients = this.syncingClients.filter(function (cli) {
}) return cli !== user
for (var f of this.userEventListeners) {
f({
action: 'userLeft',
user: user
}) })
for (var f of this.userEventListeners) {
f({
action: 'userLeft',
user: user
})
}
} }
} }
userJoined (user, role) { userJoined (user, role) {
@ -162,6 +169,34 @@ module.exports = function (Y/* :any */) {
console.log(`send ${this.userId} -> ${uid}: ${message.type}`, message) // eslint-disable-line console.log(`send ${this.userId} -> ${uid}: ${message.type}`, message) // eslint-disable-line
} }
} }
/*
Buffer operations, and broadcast them when ready.
*/
broadcastOps (ops) {
ops = ops.map(function (op) {
return Y.Struct[op.struct].encode(op)
})
var self = this
function broadcastOperations () {
if (self.broadcastOpBuffer.length > 0) {
self.broadcast({
type: 'update',
ops: self.broadcastOpBuffer
})
self.broadcastOpBuffer = []
}
}
if (this.broadcastOpBuffer.length === 0) {
this.broadcastOpBuffer = ops
if (this.y.db.transactionInProgress) {
this.y.db.whenTransactionsFinished().then(broadcastOperations)
} else {
setTimeout(broadcastOperations, 0)
}
} else {
this.broadcastOpBuffer = this.broadcastOpBuffer.concat(ops)
}
}
/* /*
You received a raw message, and you know that it is intended for Yjs. Then call this function. You received a raw message, and you know that it is intended for Yjs. Then call this function.
*/ */
@ -222,15 +257,14 @@ module.exports = function (Y/* :any */) {
db.requestTransaction(function * () { db.requestTransaction(function * () {
var ops = yield* this.getOperations(m.stateSet) var ops = yield* this.getOperations(m.stateSet)
if (ops.length > 0) { if (ops.length > 0) {
var update /* :MessageUpdate */ = {
type: 'update',
ops: ops
}
if (!broadcastHB) { // TODO: consider to broadcast here.. if (!broadcastHB) { // TODO: consider to broadcast here..
conn.send(sender, update) conn.send(sender, {
type: 'update',
ops: ops
})
} else { } else {
// broadcast only once! // broadcast only once!
conn.broadcast(update) conn.broadcastOps(ops)
} }
} }
defer.resolve() defer.resolve()
@ -252,10 +286,7 @@ module.exports = function (Y/* :any */) {
return o.struct === 'Delete' return o.struct === 'Delete'
}) })
if (delops.length > 0) { if (delops.length > 0) {
this.broadcast({ this.broadcastOps(delops)
type: 'update',
ops: delops
})
} }
} }
this.y.db.apply(message.ops) this.y.db.apply(message.ops)

View File

@ -39,6 +39,13 @@ module.exports = function (Y /* :any */) {
*/ */
constructor (y, opts) { constructor (y, opts) {
this.y = y this.y = y
var os = this
this.userId = null
var resolve
this.userIdPromise = new Promise(function (r) {
resolve = r
})
this.userIdPromise.resolve = resolve
// whether to broadcast all applied operations (insert & delete hook) // whether to broadcast all applied operations (insert & delete hook)
this.forwardAppliedOperations = false this.forwardAppliedOperations = false
// E.g. this.listenersById[id] : Array<Listener> // E.g. this.listenersById[id] : Array<Listener>
@ -60,32 +67,42 @@ module.exports = function (Y /* :any */) {
// TODO: Use ES7 Weak Maps. This way types that are no longer user, // TODO: Use ES7 Weak Maps. This way types that are no longer user,
// wont be kept in memory. // wont be kept in memory.
this.initializedTypes = {} this.initializedTypes = {}
this.whenUserIdSetListener = null
this.waitingTransactions = [] this.waitingTransactions = []
this.transactionInProgress = false this.transactionInProgress = false
this.transactionIsFlushed = false
if (typeof YConcurrency_TestingMode !== 'undefined') { if (typeof YConcurrency_TestingMode !== 'undefined') {
this.executeOrder = [] this.executeOrder = []
} }
this.gc1 = [] // first stage this.gc1 = [] // first stage
this.gc2 = [] // second stage -> after that, remove the op this.gc2 = [] // second stage -> after that, remove the op
this.gcTimeout = opts.gcTimeout || 5000 this.gcTimeout = opts.gcTimeout || 50000
var os = this
function garbageCollect () { function garbageCollect () {
return new Promise((resolve) => { return os.whenTransactionsFinished().then(function () {
os.requestTransaction(function * () { if (os.gc1.length > 0 || os.gc2.length > 0) {
if (os.y.connector != null && os.y.connector.isSynced) { return new Promise((resolve) => {
for (var i = 0; i < os.gc2.length; i++) { os.requestTransaction(function * () {
var oid = os.gc2[i] if (os.y.connector != null && os.y.connector.isSynced) {
yield* this.garbageCollectOperation(oid) for (var i = 0; i < os.gc2.length; i++) {
} var oid = os.gc2[i]
os.gc2 = os.gc1 yield* this.garbageCollectOperation(oid)
os.gc1 = [] }
} os.gc2 = os.gc1
os.gc1 = []
}
// TODO: Use setInterval here instead (when garbageCollect is called several times there will be several timeouts..)
if (os.gcTimeout > 0) {
os.gcInterval = setTimeout(garbageCollect, os.gcTimeout)
}
resolve()
})
})
} else {
// TODO: see above
if (os.gcTimeout > 0) { if (os.gcTimeout > 0) {
os.gcInterval = setTimeout(garbageCollect, os.gcTimeout) os.gcInterval = setTimeout(garbageCollect, os.gcTimeout)
} }
resolve() return Promise.resolve()
}) }
}) })
} }
this.garbageCollect = garbageCollect this.garbageCollect = garbageCollect
@ -164,26 +181,20 @@ module.exports = function (Y /* :any */) {
this.gcInterval = null this.gcInterval = null
} }
setUserId (userId) { setUserId (userId) {
var self = this if (!this.userIdPromise.inProgress) {
return new Promise(function (resolve) { this.userIdPromise.inProgress = true
var self = this
self.requestTransaction(function * () { self.requestTransaction(function * () {
self.userId = userId self.userId = userId
var state = yield* this.getState(userId) var state = yield* this.getState(userId)
self.opClock = state.clock self.opClock = state.clock
if (self.whenUserIdSetListener != null) { self.userIdPromise.resolve(userId)
self.whenUserIdSetListener()
self.whenUserIdSetListener = null
}
resolve()
}) })
}) }
return this.userIdPromise
} }
whenUserIdSet (f) { whenUserIdSet (f) {
if (this.userId != null) { this.userIdPromise.then(f)
f()
} else {
this.whenUserIdSetListener = f
}
} }
getNextOpId () { getNextOpId () {
if (this._nextUserId != null) { if (this._nextUserId != null) {
@ -205,8 +216,10 @@ module.exports = function (Y /* :any */) {
apply (ops) { apply (ops) {
for (var key in ops) { for (var key in ops) {
var o = ops[key] var o = ops[key]
var required = Y.Struct[o.struct].requiredOps(o) if (o.id == null || o.id[0] !== this.y.connector.userId) {
this.whenOperationsExist(required, o) var required = Y.Struct[o.struct].requiredOps(o)
this.whenOperationsExist(required, o)
}
} }
} }
/* /*
@ -352,7 +365,7 @@ module.exports = function (Y /* :any */) {
yield* Y.Struct['Delete'].execute.call(transaction, delop) yield* Y.Struct['Delete'].execute.call(transaction, delop)
} }
// notify parent, if it has been initialized as a custom type // notify parent, if it was instanciated as a custom type
if (t != null) { if (t != null) {
yield* t._changed(transaction, Y.utils.copyObject(op)) yield* t._changed(transaction, Y.utils.copyObject(op))
} }
@ -377,29 +390,36 @@ module.exports = function (Y /* :any */) {
return Promise.resolve() return Promise.resolve()
} }
} }
// Check if there is another transaction request.
// * the last transaction is always a flush :)
getNextRequest () { getNextRequest () {
if (this.waitingTransactions.length === 0) { if (this.waitingTransactions.length === 0) {
this.transactionInProgress = false if (this.transactionIsFlushed) {
if (this.transactionsFinished != null) { this.transactionInProgress = false
this.transactionsFinished.resolve() this.transactionIsFlushed = false
this.transactionsFinished = null if (this.transactionsFinished != null) {
this.transactionsFinished.resolve()
this.transactionsFinished = null
}
return null
} else {
this.transactionIsFlushed = true
return function * () {
yield* this.flush()
}
} }
return null
} else { } else {
this.transactionIsFlushed = false
return this.waitingTransactions.shift() return this.waitingTransactions.shift()
} }
} }
requestTransaction (makeGen/* :any */, callImmediately) { requestTransaction (makeGen/* :any */, callImmediately) {
if (true || callImmediately) { // TODO: decide whether this is ok or not.. this.waitingTransactions.push(makeGen)
this.waitingTransactions.push(makeGen) if (!this.transactionInProgress) {
if (!this.transactionInProgress) { this.transactionInProgress = true
this.transactionInProgress = true if (false || callImmediately) { // TODO: decide whether this is ok or not..
this.transact(this.getNextRequest()) this.transact(this.getNextRequest())
} } else {
} else {
this.waitingTransactions.push(makeGen)
if (!this.transactionInProgress) {
this.transactionInProgress = true
var self = this var self = this
setTimeout(function () { setTimeout(function () {
self.transact(self.getNextRequest()) self.transact(self.getNextRequest())

View File

@ -23,7 +23,7 @@ for (let database of databases) {
}) })
it('Deleted operation is deleted', async(function * (done) { it('Deleted operation is deleted', async(function * (done) {
store.requestTransaction(function * () { store.requestTransaction(function * () {
yield* this.markDeleted(['u1', 10]) yield* this.markDeleted(['u1', 10], 1)
expect(yield* this.isDeleted(['u1', 10])).toBeTruthy() expect(yield* this.isDeleted(['u1', 10])).toBeTruthy()
expect(yield* this.getDeleteSet()).toEqual({'u1': [[10, 1, false]]}) expect(yield* this.getDeleteSet()).toEqual({'u1': [[10, 1, false]]})
done() done()
@ -31,8 +31,8 @@ for (let database of databases) {
})) }))
it('Deleted operation extends other deleted operation', async(function * (done) { it('Deleted operation extends other deleted operation', async(function * (done) {
store.requestTransaction(function * () { store.requestTransaction(function * () {
yield* this.markDeleted(['u1', 10]) yield* this.markDeleted(['u1', 10], 1)
yield* this.markDeleted(['u1', 11]) yield* this.markDeleted(['u1', 11], 1)
expect(yield* this.isDeleted(['u1', 10])).toBeTruthy() expect(yield* this.isDeleted(['u1', 10])).toBeTruthy()
expect(yield* this.isDeleted(['u1', 11])).toBeTruthy() expect(yield* this.isDeleted(['u1', 11])).toBeTruthy()
expect(yield* this.getDeleteSet()).toEqual({'u1': [[10, 2, false]]}) expect(yield* this.getDeleteSet()).toEqual({'u1': [[10, 2, false]]})
@ -41,77 +41,77 @@ for (let database of databases) {
})) }))
it('Deleted operation extends other deleted operation', async(function * (done) { it('Deleted operation extends other deleted operation', async(function * (done) {
store.requestTransaction(function * () { store.requestTransaction(function * () {
yield* this.markDeleted(['0', 3]) yield* this.markDeleted(['0', 3], 1)
yield* this.markDeleted(['0', 4]) yield* this.markDeleted(['0', 4], 1)
yield* this.markDeleted(['0', 2]) yield* this.markDeleted(['0', 2], 1)
expect(yield* this.getDeleteSet()).toEqual({'0': [[2, 3, false]]}) expect(yield* this.getDeleteSet()).toEqual({'0': [[2, 3, false]]})
done() done()
}) })
})) }))
it('Debug #1', async(function * (done) { it('Debug #1', async(function * (done) {
store.requestTransaction(function * () { store.requestTransaction(function * () {
yield* this.markDeleted(['166', 0]) yield* this.markDeleted(['166', 0], 1)
yield* this.markDeleted(['166', 2]) yield* this.markDeleted(['166', 2], 1)
yield* this.markDeleted(['166', 0]) yield* this.markDeleted(['166', 0], 1)
yield* this.markDeleted(['166', 2]) yield* this.markDeleted(['166', 2], 1)
yield* this.markGarbageCollected(['166', 2]) yield* this.markGarbageCollected(['166', 2], 1)
yield* this.markDeleted(['166', 1]) yield* this.markDeleted(['166', 1], 1)
yield* this.markDeleted(['166', 3]) yield* this.markDeleted(['166', 3], 1)
yield* this.markGarbageCollected(['166', 3]) yield* this.markGarbageCollected(['166', 3], 1)
yield* this.markDeleted(['166', 0]) yield* this.markDeleted(['166', 0], 1)
expect(yield* this.getDeleteSet()).toEqual({'166': [[0, 2, false], [2, 2, true]]}) expect(yield* this.getDeleteSet()).toEqual({'166': [[0, 2, false], [2, 2, true]]})
done() done()
}) })
})) }))
it('Debug #2', async(function * (done) { it('Debug #2', async(function * (done) {
store.requestTransaction(function * () { store.requestTransaction(function * () {
yield* this.markDeleted(['293', 0]) yield* this.markDeleted(['293', 0], 1)
yield* this.markDeleted(['291', 2]) yield* this.markDeleted(['291', 2], 1)
yield* this.markDeleted(['291', 2]) yield* this.markDeleted(['291', 2], 1)
yield* this.markGarbageCollected(['293', 0]) yield* this.markGarbageCollected(['293', 0], 1)
yield* this.markDeleted(['293', 1]) yield* this.markDeleted(['293', 1], 1)
yield* this.markGarbageCollected(['291', 2]) yield* this.markGarbageCollected(['291', 2], 1)
expect(yield* this.getDeleteSet()).toEqual({'291': [[2, 1, true]], '293': [[0, 1, true], [1, 1, false]]}) expect(yield* this.getDeleteSet()).toEqual({'291': [[2, 1, true]], '293': [[0, 1, true], [1, 1, false]]})
done() done()
}) })
})) }))
it('Debug #3', async(function * (done) { it('Debug #3', async(function * (done) {
store.requestTransaction(function * () { store.requestTransaction(function * () {
yield* this.markDeleted(['581', 0]) yield* this.markDeleted(['581', 0], 1)
yield* this.markDeleted(['581', 1]) yield* this.markDeleted(['581', 1], 1)
yield* this.markDeleted(['580', 0]) yield* this.markDeleted(['580', 0], 1)
yield* this.markDeleted(['580', 0]) yield* this.markDeleted(['580', 0], 1)
yield* this.markGarbageCollected(['581', 0]) yield* this.markGarbageCollected(['581', 0], 1)
yield* this.markDeleted(['581', 2]) yield* this.markDeleted(['581', 2], 1)
yield* this.markDeleted(['580', 1]) yield* this.markDeleted(['580', 1], 1)
yield* this.markDeleted(['580', 2]) yield* this.markDeleted(['580', 2], 1)
yield* this.markDeleted(['580', 1]) yield* this.markDeleted(['580', 1], 1)
yield* this.markDeleted(['580', 2]) yield* this.markDeleted(['580', 2], 1)
yield* this.markGarbageCollected(['581', 2]) yield* this.markGarbageCollected(['581', 2], 1)
yield* this.markGarbageCollected(['581', 1]) yield* this.markGarbageCollected(['581', 1], 1)
yield* this.markGarbageCollected(['580', 1]) yield* this.markGarbageCollected(['580', 1], 1)
expect(yield* this.getDeleteSet()).toEqual({'580': [[0, 1, false], [1, 1, true], [2, 1, false]], '581': [[0, 3, true]]}) expect(yield* this.getDeleteSet()).toEqual({'580': [[0, 1, false], [1, 1, true], [2, 1, false]], '581': [[0, 3, true]]})
done() done()
}) })
})) }))
it('Debug #4', async(function * (done) { it('Debug #4', async(function * (done) {
store.requestTransaction(function * () { store.requestTransaction(function * () {
yield* this.markDeleted(['544', 0]) yield* this.markDeleted(['544', 0], 1)
yield* this.markDeleted(['543', 2]) yield* this.markDeleted(['543', 2], 1)
yield* this.markDeleted(['544', 0]) yield* this.markDeleted(['544', 0], 1)
yield* this.markDeleted(['543', 2]) yield* this.markDeleted(['543', 2], 1)
yield* this.markGarbageCollected(['544', 0]) yield* this.markGarbageCollected(['544', 0], 1)
yield* this.markDeleted(['545', 1]) yield* this.markDeleted(['545', 1], 1)
yield* this.markDeleted(['543', 4]) yield* this.markDeleted(['543', 4], 1)
yield* this.markDeleted(['543', 3]) yield* this.markDeleted(['543', 3], 1)
yield* this.markDeleted(['544', 1]) yield* this.markDeleted(['544', 1], 1)
yield* this.markDeleted(['544', 2]) yield* this.markDeleted(['544', 2], 1)
yield* this.markDeleted(['544', 1]) yield* this.markDeleted(['544', 1], 1)
yield* this.markDeleted(['544', 2]) yield* this.markDeleted(['544', 2], 1)
yield* this.markGarbageCollected(['543', 2]) yield* this.markGarbageCollected(['543', 2], 1)
yield* this.markGarbageCollected(['543', 4]) yield* this.markGarbageCollected(['543', 4], 1)
yield* this.markGarbageCollected(['544', 2]) yield* this.markGarbageCollected(['544', 2], 1)
yield* this.markGarbageCollected(['543', 3]) yield* this.markGarbageCollected(['543', 3], 1)
expect(yield* this.getDeleteSet()).toEqual({'543': [[2, 3, true]], '544': [[0, 1, true], [1, 1, false], [2, 1, true]], '545': [[1, 1, false]]}) expect(yield* this.getDeleteSet()).toEqual({'543': [[2, 3, true]], '544': [[0, 1, true], [1, 1, false], [2, 1, true]], '545': [[1, 1, false]]})
done() done()
}) })
@ -136,21 +136,21 @@ for (let database of databases) {
})) }))
it('Debug #7', async(function * (done) { it('Debug #7', async(function * (done) {
store.requestTransaction(function * () { store.requestTransaction(function * () {
yield* this.markDeleted(['9', 2]) yield* this.markDeleted(['9', 2], 1)
yield* this.markDeleted(['11', 2]) yield* this.markDeleted(['11', 2], 1)
yield* this.markDeleted(['11', 4]) yield* this.markDeleted(['11', 4], 1)
yield* this.markDeleted(['11', 1]) yield* this.markDeleted(['11', 1], 1)
yield* this.markDeleted(['9', 4]) yield* this.markDeleted(['9', 4], 1)
yield* this.markDeleted(['10', 0]) yield* this.markDeleted(['10', 0], 1)
yield* this.markGarbageCollected(['11', 2]) yield* this.markGarbageCollected(['11', 2], 1)
yield* this.markDeleted(['11', 2]) yield* this.markDeleted(['11', 2], 1)
yield* this.markGarbageCollected(['11', 3]) yield* this.markGarbageCollected(['11', 3], 1)
yield* this.markDeleted(['11', 3]) yield* this.markDeleted(['11', 3], 1)
yield* this.markDeleted(['11', 3]) yield* this.markDeleted(['11', 3], 1)
yield* this.markDeleted(['9', 4]) yield* this.markDeleted(['9', 4], 1)
yield* this.markDeleted(['10', 0]) yield* this.markDeleted(['10', 0], 1)
yield* this.markGarbageCollected(['11', 1]) yield* this.markGarbageCollected(['11', 1], 1)
yield* this.markDeleted(['11', 1]) yield* this.markDeleted(['11', 1], 1)
expect(yield* this.getDeleteSet()).toEqual({'9': [[2, 1, false], [4, 1, false]], '10': [[0, 1, false]], '11': [[1, 3, true], [4, 1, false]]}) expect(yield* this.getDeleteSet()).toEqual({'9': [[2, 1, false], [4, 1, false]], '10': [[0, 1, false]], '11': [[1, 3, true], [4, 1, false]]})
done() done()
}) })

View File

@ -25,7 +25,7 @@ g.g = g
g.YConcurrency_TestingMode = true g.YConcurrency_TestingMode = true
jasmine.DEFAULT_TIMEOUT_INTERVAL = 8000 jasmine.DEFAULT_TIMEOUT_INTERVAL = 10000
g.describeManyTimes = function describeManyTimes (times, name, f) { g.describeManyTimes = function describeManyTimes (times, name, f) {
for (var i = 0; i < times; i++) { for (var i = 0; i < times; i++) {
@ -77,6 +77,12 @@ function getRandomNumber (n) {
} }
g.getRandomNumber = getRandomNumber g.getRandomNumber = getRandomNumber
function getRandomString () {
var tokens = 'abcdefäö' // ü\n\n\n\n\n\n\n'
return tokens[getRandomNumber(tokens.length - 1)]
}
g.getRandomString = getRandomString
function * applyTransactions (relAmount, numberOfTransactions, objects, users, transactions) { function * applyTransactions (relAmount, numberOfTransactions, objects, users, transactions) {
function randomTransaction (root) { function randomTransaction (root) {
var f = getRandom(transactions) var f = getRandom(transactions)
@ -202,6 +208,7 @@ g.compareAllUsers = async(function * compareAllUsers (users) {
yield* this.os.iterate(this, null, null, function * (o) { yield* this.os.iterate(this, null, null, function * (o) {
o = Y.utils.copyObject(o) o = Y.utils.copyObject(o)
delete o.origin delete o.origin
delete o.originOf
db1.push(o) db1.push(o)
}) })
}) })
@ -216,6 +223,7 @@ g.compareAllUsers = async(function * compareAllUsers (users) {
yield* this.os.iterate(this, null, null, function * (o) { yield* this.os.iterate(this, null, null, function * (o) {
o = Y.utils.copyObject(o) o = Y.utils.copyObject(o)
delete o.origin delete o.origin
delete o.originOf
expect(db1[count++]).toEqual(o) expect(db1[count++]).toEqual(o)
}) })
}) })

View File

@ -126,6 +126,19 @@ module.exports = function (Y/* :any */) {
execute: function *(op) { execute: function *(op) {
var i // loop counter var i // loop counter
var distanceToOrigin = i = yield* Struct.Insert.getDistanceToOrigin.call(this, op) // most cases: 0 (starts from 0) var distanceToOrigin = i = yield* Struct.Insert.getDistanceToOrigin.call(this, op) // most cases: 0 (starts from 0)
if (op.origin != null) { // TODO: !== instead of !=
// we save in origin that op originates in it
// we need that later when we eventually garbage collect origin (see transaction)
var origin = yield* this.getOperation(op.origin)
if (origin.originOf == null) {
origin.originOf = []
}
origin.originOf.push(op.id)
yield* this.setOperation(origin)
}
// now we begin to insert op in the list of insertions..
var o var o
var parent var parent
var start var start

View File

@ -123,10 +123,7 @@ module.exports = function (Y/* :any */) {
} }
if (!this.store.y.connector.isDisconnected() && send.length > 0) { // TODO: && !this.store.forwardAppliedOperations (but then i don't send delete ops) if (!this.store.y.connector.isDisconnected() && send.length > 0) { // TODO: && !this.store.forwardAppliedOperations (but then i don't send delete ops)
// is connected, and this is not going to be send in addOperation // is connected, and this is not going to be send in addOperation
this.store.y.connector.broadcast({ this.store.y.connector.broadcastOps(send)
type: 'update',
ops: send
})
} }
} }
@ -153,7 +150,7 @@ module.exports = function (Y/* :any */) {
var callType = false var callType = false
if (target == null || !target.deleted) { if (target == null || !target.deleted) {
yield* this.markDeleted(targetId) yield* this.markDeleted(targetId, 1)
} }
if (target != null && target.gc == null) { if (target != null && target.gc == null) {
@ -225,86 +222,124 @@ module.exports = function (Y/* :any */) {
/* /*
Mark an operation as deleted&gc'd Mark an operation as deleted&gc'd
*/ */
* markGarbageCollected (id) { * markGarbageCollected (id, len) {
// this.mem.push(["gc", id]); // this.mem.push(["gc", id]);
var n = yield* this.markDeleted(id) var n = yield* this.markDeleted(id, len)
if (!n.gc) { if (n.id[1] < id[1] && !n.gc) {
if (n.id[1] < id[1]) { // un-extend left
// un-extend left var newlen = n.len - (id[1] - n.id[1])
var newlen = n.len - (id[1] - n.id[1]) n.len -= newlen
n.len -= newlen yield* this.ds.put(n)
yield* this.ds.put(n) n = {id: id, len: newlen, gc: false}
n = {id: id, len: newlen, gc: false}
yield* this.ds.put(n)
}
// get prev&next before adding a new operation
var prev = yield* this.ds.findPrev(id)
var next = yield* this.ds.findNext(id)
if (id[1] < n.id[1] + n.len - 1) {
// un-extend right
yield* this.ds.put({id: [id[0], id[1] + 1], len: n.len - 1, gc: false})
n.len = 1
}
// set gc'd
n.gc = true
// can extend left?
if (
prev != null &&
prev.gc &&
Y.utils.compareIds([prev.id[0], prev.id[1] + prev.len], n.id)
) {
prev.len += n.len
yield* this.ds.delete(n.id)
n = prev
// ds.put n here?
}
// can extend right?
if (
next != null &&
next.gc &&
Y.utils.compareIds([n.id[0], n.id[1] + n.len], next.id)
) {
n.len += next.len
yield* this.ds.delete(next.id)
}
yield* this.ds.put(n) yield* this.ds.put(n)
} }
// get prev&next before adding a new operation
var prev = yield* this.ds.findPrev(id)
var next = yield* this.ds.findNext(id)
if (id[1] < n.id[1] + n.len - len && !n.gc) {
// un-extend right
yield* this.ds.put({id: [id[0], id[1] + 1], len: n.len - 1, gc: false})
n.len = 1
}
// set gc'd
n.gc = true
// can extend left?
if (
prev != null &&
prev.gc &&
Y.utils.compareIds([prev.id[0], prev.id[1] + prev.len], n.id)
) {
prev.len += n.len
yield* this.ds.delete(n.id)
n = prev
// ds.put n here?
}
// can extend right?
if (
next != null &&
next.gc &&
Y.utils.compareIds([n.id[0], n.id[1] + n.len], next.id)
) {
n.len += next.len
yield* this.ds.delete(next.id)
}
yield* this.ds.put(n)
} }
/* /*
Mark an operation as deleted. Mark an operation as deleted.
returns the delete node returns the delete node
*/ */
* markDeleted (id) { * markDeleted (id, length) {
if (length == null) {
length = 1
}
// this.mem.push(["del", id]); // this.mem.push(["del", id]);
var n = yield* this.ds.findWithUpperBound(id) var n = yield* this.ds.findWithUpperBound(id)
if (n != null && n.id[0] === id[0]) { if (n != null && n.id[0] === id[0]) {
if (n.id[1] <= id[1] && id[1] < n.id[1] + n.len) { if (n.id[1] <= id[1] && id[1] <= n.id[1] + n.len) {
// already deleted // id is in n's range
return n var diff = id[1] + length - (n.id[1] + n.len) // overlapping right
} else if (n.id[1] + n.len === id[1] && !n.gc) { if (diff > 0) {
// can extend existing deletion // id+length overlaps n
n.len++ if (!n.gc) {
n.len += diff
} else {
diff = n.id[1] + n.len - id[1] // overlapping left (id till n.end)
if (diff < length) {
// a partial deletion
n = {id: [id[0], id[1] + diff], len: length - diff, gc: false}
yield* this.ds.put(n)
} else {
// already gc'd
throw new Error('Cannot happen! (it dit though.. :()')
// return n
}
}
} else {
// no overlapping, already deleted
return n
}
} else { } else {
// cannot extend left // cannot extend left (there is no left!)
n = {id: id, len: 1, gc: false} n = {id: id, len: length, gc: false}
yield* this.ds.put(n) yield* this.ds.put(n) // TODO: you double-put !!
} }
} else { } else {
// cannot extend left // cannot extend left
n = {id: id, len: 1, gc: false} n = {id: id, len: length, gc: false}
yield* this.ds.put(n) yield* this.ds.put(n)
} }
// can extend right? // can extend right?
var next = yield* this.ds.findNext(n.id) var next = yield* this.ds.findNext(n.id)
if ( if (
next != null && next != null &&
Y.utils.compareIds([n.id[0], n.id[1] + n.len], next.id) && n.id[0] === next.id[0] &&
!next.gc n.id[1] + n.len >= next.id[1]
) { ) {
n.len = n.len + next.len diff = n.id[1] + n.len - next.id[1] // from next.start to n.end
yield* this.ds.delete(next.id) if (next.gc) {
if (diff >= 0) {
n.len -= diff
if (diff > next.len) {
// need to create another deletion after $next
// TODO: (may not be necessary, because this case shouldn't happen!)
// also this is supposed to return a deletion range. which one to choose? n or the new created deletion?
throw new Error('This case is not handled (on purpose!)')
}
} // else: everything is fine :)
} else {
if (diff >= 0) {
if (diff > next.len) {
// may be neccessary to extend next.next!
// TODO: (may not be necessary, because this case shouldn't happen!)
throw new Error('This case is not handled (on purpose!)')
}
n.len += next.len - diff
yield* this.ds.delete(next.id)
}
}
} }
yield* this.ds.put(n) yield* this.ds.put(n)
return n return n
@ -333,19 +368,9 @@ module.exports = function (Y/* :any */) {
*/ */
* garbageCollectOperation (id) { * garbageCollectOperation (id) {
this.store.addToDebug('yield* this.garbageCollectOperation(', id, ')') this.store.addToDebug('yield* this.garbageCollectOperation(', id, ')')
// check to increase the state of the respective user
var state = yield* this.getState(id[0])
if (state.clock === id[1]) {
state.clock++
// also check if more expected operations were gc'd
yield* this.checkDeleteStoreForState(state)
// then set the state
yield* this.setState(state)
}
yield* this.markGarbageCollected(id)
// if op exists, then clean that mess up..
var o = yield* this.getOperation(id) var o = yield* this.getOperation(id)
yield* this.markGarbageCollected(id, 1) // always mark gc'd
// if op exists, then clean that mess up..
if (o != null) { if (o != null) {
/* /*
if (!o.deleted) { if (!o.deleted) {
@ -365,23 +390,31 @@ module.exports = function (Y/* :any */) {
if (o.right != null) { if (o.right != null) {
var right = yield* this.getOperation(o.right) var right = yield* this.getOperation(o.right)
right.left = o.left right.left = o.left
if (Y.utils.compareIds(right.origin, o.id)) { // rights origin is o
if (o.originOf != null && o.originOf.length > 0) {
// find new origin of right ops // find new origin of right ops
// origin is the first left deleted operation // origin is the first left deleted operation
var neworigin = o.left var neworigin = o.left
var neworigin_ = null
while (neworigin != null) { while (neworigin != null) {
var neworigin_ = yield* this.getOperation(neworigin) neworigin_ = yield* this.getOperation(neworigin)
if (neworigin_.deleted) { if (neworigin_.deleted) {
break break
} }
neworigin = neworigin_.left neworigin = neworigin_.left
} }
// reset origin of all right ops (except first right - duh!),
/* ** The following code does not rely on the the originOf property **
I recently added originOf to all Insert Operations (see Struct.Insert.execute),
which saves which operations originate in a Insert operation.
Garbage collecting without originOf is more memory efficient, but is nearly impossible for large texts, or lists!
But I keep this code for now
```
// reset origin of right // reset origin of right
right.origin = neworigin right.origin = neworigin
// search until you find origin pointer to the left of o
// reset origin of all right ops (except first right - duh!),
// until you find origin pointer to the left of o
if (right.right != null) { if (right.right != null) {
var i = yield* this.getOperation(right.right) var i = yield* this.getOperation(right.right)
var ids = [o.id, o.right] var ids = [o.id, o.right]
@ -397,13 +430,46 @@ module.exports = function (Y/* :any */) {
if (i.right == null) { if (i.right == null) {
break break
} else { } else {
ids.push(i.id)
i = yield* this.getOperation(i.right) i = 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) // ** Now the new implementation starts **
// reset neworigin of all originOf[*]
for (var _i in o.originOf) {
var originsIn = yield* this.getOperation(o.originOf[_i])
if (originsIn != null) {
originsIn.origin = neworigin
yield* this.setOperation(originsIn)
}
}
if (neworigin != null) {
if (neworigin_.originOf == null) {
neworigin_.originOf = o.originOf
} else {
neworigin_.originOf = o.originOf.concat(neworigin_.originOf)
}
yield* this.setOperation(neworigin_)
}
// we don't need to set right here, because
// right should be in o.originOf => it is set it the previous for loop
} else {
// we didn't need to reset the origin of right
// so we have to set right here
yield* this.setOperation(right)
}
// o may originate in another operation.
// Since o is deleted, we have to reset o.origin's `originOf` property
if (o.origin != null) {
var origin = yield* this.getOperation(o.origin)
origin.originOf = origin.originOf.filter(function (_id) {
return !Y.utils.compareIds(id, _id)
})
yield* this.setOperation(origin)
}
} }
if (o.parent != null) { if (o.parent != null) {
@ -448,9 +514,7 @@ module.exports = function (Y/* :any */) {
* applyDeleteSet (ds) { * applyDeleteSet (ds) {
var deletions = [] var deletions = []
function createDeletions (user, start, len, gc) { function createDeletions (user, start, len, gc) {
for (var c = start; c < start + len; c++) { deletions.push([user, start, len, gc])
deletions.push([user, c, gc])
}
} }
for (var user in ds) { for (var user in ds) {
@ -505,26 +569,45 @@ module.exports = function (Y/* :any */) {
} }
for (var i = 0; i < deletions.length; i++) { for (var i = 0; i < deletions.length; i++) {
var del = deletions[i] var del = deletions[i]
var id = [del[0], del[1]]
// always try to delete.. // always try to delete..
var addOperation = yield* this.deleteOperation(id) var state = yield* this.getState(del[0])
if (addOperation) { if (del[1] < state.clock) {
// TODO:.. really .. here? You could prevent calling all these functions in operationAdded for (let c = del[1]; c < del[1] + del[2]; c++) {
yield* this.store.operationAdded(this, {struct: 'Delete', target: id}) var id = [del[0], c]
var addOperation = yield* this.deleteOperation(id)
if (addOperation) {
// TODO:.. really .. here? You could prevent calling all these functions in operationAdded
yield* this.store.operationAdded(this, {struct: 'Delete', target: id})
}
if (del[3]) {
// gc
yield* this.garbageCollectOperation(id)
}
}
} else {
if (del[3]) {
yield* this.markGarbageCollected([del[0], del[1]], del[2])
} else {
yield* this.markDeleted([del[0], del[1]], del[2])
}
} }
if (del[2]) { if (del[3]) {
// gc // check to increase the state of the respective user
yield* this.garbageCollectOperation(id) if (state.clock >= del[1] && state.clock < del[1] + del[2]) {
state.clock = del[1] + del[2]
// also check if more expected operations were gc'd
yield* this.checkDeleteStoreForState(state) // TODO: unneccessary?
// then set the state
yield* this.setState(state)
}
}
if (this.store.forwardAppliedOperations) {
var ops = []
for (let c = del[1]; c < del[1] + del[2]; c++) {
ops.push({struct: 'Delete', target: [d[0], c]}) // TODO: implement Delete with deletion length!
}
this.store.y.connector.broadcastOps(ops)
} }
}
if (this.store.forwardAppliedOperations) {
var ops = deletions.map(function (d) {
return {struct: 'Delete', target: [d[0], d[1]]}
})
this.store.y.connector.broadcast({
type: 'update',
ops: ops
})
} }
} }
* isGarbageCollected (id) { * isGarbageCollected (id) {
@ -562,10 +645,7 @@ module.exports = function (Y/* :any */) {
yield* this.os.put(op) yield* this.os.put(op)
if (!this.store.y.connector.isDisconnected() && this.store.forwardAppliedOperations && op.id[0] !== '_') { if (!this.store.y.connector.isDisconnected() && this.store.forwardAppliedOperations && op.id[0] !== '_') {
// is connected, and this is not going to be send in addOperation // is connected, and this is not going to be send in addOperation
this.store.y.connector.broadcast({ this.store.y.connector.broadcastOps([op])
type: 'update',
ops: [op]
})
} }
} }
* getOperation (id/* :any */)/* :Transaction<any> */ { * getOperation (id/* :any */)/* :Transaction<any> */ {
@ -625,34 +705,8 @@ module.exports = function (Y/* :any */) {
}) })
return ss return ss
} }
* getOperations (startSS) {
// TODO: use bounds here!
if (startSS == null) {
startSS = {}
}
var ops = []
var endSV = yield* this.getStateVector()
for (var endState of endSV) {
var user = endState.user
if (user === '_') {
continue
}
var startPos = startSS[user] || 0
yield* this.os.iterate(this, [user, startPos], [user, Number.MAX_VALUE], function * (op) {
ops.push(op)
})
}
var res = []
for (var op of ops) {
var o = yield* this.makeOperationReady(startSS, op)
res.push(o)
}
return res
}
/* /*
Here, we make op executable for the receiving user. Here, we make all missing operations executable for the receiving user.
Notes: Notes:
startSS: denotes to the SV that the remote user sent startSS: denotes to the SV that the remote user sent
@ -687,7 +741,92 @@ module.exports = function (Y/* :any */) {
(startSS or currSS.. ?) (startSS or currSS.. ?)
-> Could be necessary when I turn GC again. -> Could be necessary when I turn GC again.
-> Is a bad(ish) idea because it requires more computation -> Is a bad(ish) idea because it requires more computation
What we do:
* Iterate over all missing operations.
* When there is an operation, where the right op is known, send this op all missing ops to the left to the user
* I explained above what we have to do with each operation. Here is how we do it efficiently:
1. Go to the left until you find either op.origin, or a known operation (let o denote current operation in the iteration)
2. Found a known operation -> set op.left = o, and send it to the user. stop
3. Found o = op.origin -> set op.left = op.origin, and send it to the user. start again from 1. (set op = o)
4. Found some o -> set o.right = op, o.left = o.origin, send it to the user, continue
*/ */
* getOperations (startSS) {
// TODO: use bounds here!
if (startSS == null) {
startSS = {}
}
var send = []
var endSV = yield* this.getStateVector()
for (var endState of endSV) {
var user = endState.user
if (user === '_') {
continue
}
var startPos = startSS[user] || 0
yield* this.os.iterate(this, [user, startPos], [user, Number.MAX_VALUE], function * (op) {
op = Y.Struct[op.struct].encode(op)
if (op.struct !== 'Insert') {
send.push(op)
} else if (op.right == null || op.right[1] < (startSS[op.right[0]] || 0)) {
// case 1. op.right is known
var o = op
// Remember: ?
// -> set op.right
// 1. to the first operation that is known (according to startSS)
// 2. or to the first operation that has an origin that is not to the
// right of op.
// For this we maintain a list of ops which origins are not found yet.
var missing_origins = [op]
var newright = op.right
while (true) {
if (o.left == null) {
op.left = null
send.push(op)
if (!Y.utils.compareIds(o.id, op.id)) {
o = Y.Struct[op.struct].encode(o)
o.right = missing_origins[missing_origins.length - 1].id
send.push(o)
}
break
}
o = yield* this.getOperation(o.left)
// we set another o, check if we can reduce $missing_origins
while (missing_origins.length > 0 && Y.utils.compareIds(missing_origins[missing_origins.length - 1].origin, o.id)) {
missing_origins.pop()
}
if (o.id[1] < (startSS[o.id[0]] || 0)) {
// case 2. o is known
op.left = o.id
send.push(op)
break
} else if (Y.utils.compareIds(o.id, op.origin)) {
// case 3. o is op.origin
op.left = op.origin
send.push(op)
op = Y.Struct[op.struct].encode(o)
op.right = newright
if (missing_origins.length > 0) {
console.log('This should not happen .. :( please report this')
}
missing_origins = [op]
} else {
// case 4. send o, continue to find op.origin
var s = Y.Struct[op.struct].encode(o)
s.right = missing_origins[missing_origins.length - 1].id
s.left = s.origin
send.push(s)
missing_origins.push(o)
}
}
}
})
}
return send.reverse()
}
/* this is what we used before.. use this as a reference..
* makeOperationReady (startSS, op) { * makeOperationReady (startSS, op) {
op = Y.Struct[op.struct].encode(op) op = Y.Struct[op.struct].encode(op)
op = Y.utils.copyObject(op) op = Y.utils.copyObject(op)
@ -711,6 +850,12 @@ module.exports = function (Y/* :any */) {
op.left = op.origin op.left = op.origin
return op return op
} }
*/
* flush () {
yield* this.os.flush()
yield* this.ss.flush()
yield* this.ds.flush()
}
} }
Y.Transaction = TransactionInterface Y.Transaction = TransactionInterface
} }

View File

@ -20,7 +20,7 @@
``` ```
The structures usually work asynchronously (you have to wait for the The structures usually work asynchronously (you have to wait for the
database request to finish). EventHandler will help you to make your type database request to finish). EventHandler helps you to make your type
synchronous. synchronous.
*/ */
module.exports = function (Y /* : any*/) { module.exports = function (Y /* : any*/) {
@ -207,7 +207,7 @@ module.exports = function (Y /* : any*/) {
Defines a smaller relation on Id's Defines a smaller relation on Id's
*/ */
function smaller (a, b) { function smaller (a, b) {
return a[0] < b[0] || (a[0] === b[0] && a[1] < b[1]) return a[0] < b[0] || (a[0] === b[0] && (a[1] < b[1] || typeof a[1] < typeof b[1]))
} }
Y.utils.smaller = smaller Y.utils.smaller = smaller
@ -225,4 +225,157 @@ module.exports = function (Y /* : any*/) {
} }
} }
Y.utils.compareIds = compareIds Y.utils.compareIds = compareIds
function createEmptyOpsArray (n) {
var a = new Array(n)
for (var i = 0; i < a.length; i++) {
a[i] = {
id: [null, null]
}
}
return a
}
function createSmallLookupBuffer (Store) {
/*
This buffer implements a very small buffer that temporarily stores operations
after they are read / before they are written.
The buffer basically implements FIFO. Often requested lookups will be re-queued every time they are looked up / written.
It can speed up lookups on Operation Stores and State Stores. But it does not require notable use of memory or processing power.
Good for os and ss, bot not for ds (because it often uses methods that require a flush)
I tried to optimize this for performance, therefore no highlevel operations.
*/
class SmallLookupBuffer extends Store {
constructor () {
super(...arguments)
this.writeBuffer = createEmptyOpsArray(5)
this.readBuffer = createEmptyOpsArray(10)
}
* find (id) {
var i, r
for (i = this.readBuffer.length - 1; i >= 0; i--) {
r = this.readBuffer[i]
// we don't have to use compareids, because id is always defined!
if (r.id[1] === id[1] && r.id[0] === id[0]) {
// found r
// move r to the end of readBuffer
for (; i < this.readBuffer.length - 1; i++) {
this.readBuffer[i] = this.readBuffer[i + 1]
}
this.readBuffer[this.readBuffer.length - 1] = r
return r
}
}
var o
for (i = this.writeBuffer.length - 1; i >= 0; i--) {
r = this.writeBuffer[i]
if (r.id[1] === id[1] && r.id[0] === id[0]) {
o = r
break
}
}
if (i < 0) {
// did not reach break in last loop
// read id and put it to the end of readBuffer
o = yield* super.find(id)
}
if (o != null) {
for (i = 0; i < this.readBuffer.length - 1; i++) {
this.readBuffer[i] = this.readBuffer[i + 1]
}
this.readBuffer[this.readBuffer.length - 1] = o
}
return o
}
* put (o) {
var id = o.id
var i, r // helper variables
for (i = this.writeBuffer.length - 1; i >= 0; i--) {
r = this.writeBuffer[i]
if (r.id[1] === id[1] && r.id[0] === id[0]) {
// is already in buffer
// forget r, and move o to the end of writeBuffer
for (; i < this.writeBuffer.length - 1; i++) {
this.writeBuffer[i] = this.writeBuffer[i + 1]
}
this.writeBuffer[this.writeBuffer.length - 1] = o
break
}
}
if (i < 0) {
// did not reach break in last loop
// write writeBuffer[0]
var write = this.writeBuffer[0]
if (write.id[0] !== null) {
yield* super.put(write)
}
// put o to the end of writeBuffer
for (i = 0; i < this.writeBuffer.length - 1; i++) {
this.writeBuffer[i] = this.writeBuffer[i + 1]
}
this.writeBuffer[this.writeBuffer.length - 1] = o
}
// check readBuffer for every occurence of o.id, overwrite if found
// whether found or not, we'll append o to the readbuffer
for (i = 0; i < this.readBuffer.length - 1; i++) {
r = this.readBuffer[i + 1]
if (r.id[1] === id[1] && r.id[0] === id[0]) {
this.readBuffer[i] = o
} else {
this.readBuffer[i] = r
}
}
this.readBuffer[this.readBuffer.length - 1] = o
}
* delete (id) {
var i, r
for (i = 0; i < this.readBuffer.length; i++) {
r = this.readBuffer[i]
if (r.id[1] === id[1] && r.id[0] === id[0]) {
this.readBuffer[i] = {
id: [null, null]
}
}
}
yield* this.flush()
yield* super.delete(id)
}
* findWithLowerBound () {
yield* this.flush()
return yield* super.findWithLowerBound.apply(this, arguments)
}
* findWithUpperBound () {
yield* this.flush()
return yield* super.findWithUpperBound.apply(this, arguments)
}
* findNext () {
yield* this.flush()
return yield* super.findNext.apply(this, arguments)
}
* findPrev () {
yield* this.flush()
return yield* super.findPrev.apply(this, arguments)
}
* iterate () {
yield* this.flush()
yield* super.iterate.apply(this, arguments)
}
* flush () {
for (var i = 0; i < this.writeBuffer.length; i++) {
var write = this.writeBuffer[i]
if (write.id[0] !== null) {
yield* super.put(write)
this.writeBuffer[i] = {
id: [null, null]
}
}
}
}
}
return SmallLookupBuffer
}
Y.utils.createSmallLookupBuffer = createSmallLookupBuffer
} }

View File

@ -11,6 +11,7 @@ require('./Connectors/Test.js')(Y)
var requiringModules = {} var requiringModules = {}
module.exports = Y module.exports = Y
Y.requiringModules = requiringModules
Y.extend = function (name, value) { Y.extend = function (name, value) {
Y[name] = value Y[name] = value
@ -92,8 +93,9 @@ function Y (opts/* :YOptions */) /* :Promise<YConfig> */ {
Y.sourceDir = opts.sourceDir Y.sourceDir = opts.sourceDir
return Y.requestModules(modules).then(function () { return Y.requestModules(modules).then(function () {
return new Promise(function (resolve) { return new Promise(function (resolve) {
var yconfig = new YConfig(opts, function () { var yconfig = new YConfig(opts)
yconfig.db.whenUserIdSet(function () { yconfig.db.whenUserIdSet(function () {
yconfig.init(function () {
resolve(yconfig) resolve(yconfig)
}) })
}) })
@ -108,8 +110,12 @@ class YConfig {
share: {[key: string]: any}; share: {[key: string]: any};
*/ */
constructor (opts, callback) { constructor (opts, callback) {
this.options = opts
this.db = new Y[opts.db.name](this, opts.db) this.db = new Y[opts.db.name](this, opts.db)
this.connector = new Y[opts.connector.name](this, opts.connector) this.connector = new Y[opts.connector.name](this, opts.connector)
}
init (callback) {
var opts = this.options
var share = {} var share = {}
this.share = share this.share = share
this.db.requestTransaction(function * requestTransaction () { this.db.requestTransaction(function * requestTransaction () {
@ -125,7 +131,8 @@ class YConfig {
} }
share[propertyname] = yield* this.getType(id) share[propertyname] = yield* this.getType(id)
} }
setTimeout(callback, 0) this.store.whenTransactionsFinished()
.then(callback)
}) })
} }
isConnected () { isConnected () {