@hkjeffchan thanks for the feature request - can you elaborate on what you are looking for? Happy to implement as long as I understand the request.

from live-mutex.

This is a single process and callback implementation.
https://www.npmjs.com/package/locks

ReadWrite Lock:

Allow multiple concurrent read but once write lock is acquired, read and write lock request are blocked.

Basic function:
createReadWriteLock()
acquireReadLock(key)
acquireWriteLock(key)
releaseReadLock(key)
releaseWriteLock(key)

Advance function:
promoteToWriteLock(key)
demoteToReadLock(key)

Semaphore

Allow up to N instance of resource to be shared

Basic functions:
createSemaphore(resourceNum: number)
wait()
signal()

Conditional Variable

Execute when a condition was met.

Basic functions:
createCondVar(key: string, initVal: number)
updateValue(value: number) // probably a separate mutex is needed to control this
getValue(): number // probably a separate mutex is needed to control this
wait(conditionFn: (value: number) => boolean)

const mutexCondVar = new mutex('mutexCondVar');

const condVar = createCondVar('bufferLeft', 1000);
condVar.wait( (value) => value >= 750 ).then(res => console.log('buffer is almost full!'));

mutexCondVar.acquireLock().then(async res => {
  let bufferLeft = condVar.getValue('bufferLeft');
  ... // do something and update bufferLeft
  await condVar.updateValue(bufferLeft);
  mutexCondVar.releaseLock();
});

from live-mutex.

Cool - let me think about it - read/write lock makes sense and semaphore makes sense, but I don't understand the conditional variable one though - what is the purpose of the conditional variable? example use case?

from live-mutex.

I thought about it - read/write lock and semaphore can be implemented fairly easily. If you are curious, I can explain how I will implement them. However, I don't understand the "conditional variable" feature.

For read/write lock, it can be simulated like so:

// read lock requestor
const c = new Client();
c.acquire('read-key').then(v => {
    return c.release('read-key');
});

// write lock requestor
Promise.all([
  c.acquire('read-key', {force:true}), // move to front of the queue with 'force'
  c.acquire('write-key')
])
.then([a,b] => {
   // now you can do writes
});

so I do not think that the library needs to implement that. We can just document how to do a read/write lock.

For semaphores, the library has to implement something internally. We will keep a count of currently held locks - if we are less than the max, we don't queue the request. It's that simple. This can be incorporated into the existing code, where the default max is 1. After all, a mutex is basically a binary semaphore, 'semaphore(1)'.

from live-mutex.

const c = new Client();
c.acquire('read-key').then(v => {
    ... // imagine this is a long read operation here, other readers are blocked from reading
    return c.release('read-key');
});

Your implementation will only allow one process to read the resource. A read/write lock should allow multiple read. And this read/write lock is the classic use case of condition variable.

The condition variable "readerCount" must be zero for the writer to execute. The readers acquire the mutex, increment readerCount, release the mutex, then start reading. After finish reading, acquire the mutex, decrement readerCount release the mutex. So multi readers can read at the same time.

https://en.wikipedia.org/wiki/Readers%E2%80%93writer_lock

from live-mutex.

Ah yes, you are right. Let me do some more thinking.

from live-mutex.

@hkjeffchan what is your use case btw? Are you building an application or a library? What kind of application/library?

If you are building a library, you might find this useful for testing:
https://github.com/ORESoftware/r2g

from live-mutex.

So I was looking at your description of condition variable. Given your description, I cannot figure out how the writer(s) block the reader(s). If the writer just waits for readCount to be 0, it could wait forever.

Once the readCount is 0, and the writer activates, how does the writer block the readers?

from live-mutex.

I am reading about implementations - https://en.wikipedia.org/wiki/Readers%E2%80%93writer_lock

from live-mutex.

this implementation makes sense, but this implementation is read-preferring. I wonder how to create a "write-preferring" read/write lock.

Begin Read
Lock r.
Increment b.
If b = 1, lock g.
Unlock r.

End Read
Lock r.
Decrement b.
If b = 0, unlock g.
Unlock r.

Begin Write
Lock g.

End Write
Unlock g.

(This implementation is read-preferring.)

from live-mutex.

I think a write-preferring algorithm is much better. If you use a read-preferring algorithm, the writes will build-up and the readers will get old data. write-preferring is much better, as far as I can tell for most use cases.

from live-mutex.

The last implementation under https://en.wikipedia.org/wiki/Readers%E2%80%93writer_lock is write-preferring but it needs to implement the condition variable first. Once you have condition variable implemented, the code should be as simple as it describes.

I played with pthread before and it got pthread_cond_wait and pthread_cond_signal. No idea how it works internally, probably very low level system call involved.

I do implemented the read-write lock using pthread with the above algorithm before and it is really straight forward.

I am building a custom closure tree structure and want to build the read/write with the tree but it seems there is no working lib in nodejs cluster mode. Surprise!

from live-mutex.

The following is a better explanation of Condition Variable. I found it from
https://www.boost.org/doc/libs/1_47_0/doc/html/interprocess/synchronization_mechanisms.html#interprocess.synchronization_mechanisms.conditions.conditions_whats_a_condition

What's A Condition Variable?

In the previous example, a mutex is used to lock but we can't use it to wait efficiently until the condition to continue is met. A condition variable can do two things:

wait: The thread is blocked until some other thread notifies that it can continue because the condition that lead to waiting has disappeared.
notify: The thread sends a signal to one blocked thread or to all blocked threads to tell them that they the condition that provoked their wait has disappeared.

Waiting in a condition variable is always associated with a mutex. The mutex must be locked prior to waiting on the condition. When waiting on the condition variable, the thread unlocks the mutex and waits atomically.

When the thread returns from a wait function (because of a signal or a timeout, for example) the mutex object is again locked.

from live-mutex.

I started working on the semaphore - it seems to be easy to implement, just a few lines of code to change.

The conditional variable seems to be a broadcast mechanism - which is a good because since we have a client-broker model, the broker will just broadcast to the clients. The conditional variable seems to be needed for a write-preferring lock, but it doesn't seem to be necessary for a read-preferring lock.

However, to implement a write-preferring lock:
https://github.com/ORESoftware/live-mutex/blob/dev/docs/write-preferring-rw-lock.md

It looks like a writer will block readers, and readers will block readers. For every operation, the algorithm I found has to get a lock on m. Maybe there is a better algorithm for a write-preferring read/write lock?

from live-mutex.

So semaphore is implemented on the dev branch, it seems to work fine -

'use strict';

const path = require('path');
const async = require('async');
const {Broker, Client} = require('../dist');

///////////////////////////////////////////////////////////////

const conf = Object.freeze({udsPath: path.resolve(process.env.HOME + '/suman.test2.unix.sock')});

Promise.all([
  new Broker().ensure(),
  new Client().connect()
])
.then(function ([b, c]) {

  b.emitter.on('warning', function (v) {
      console.error('broker warning:',...arguments);
  });

  c.emitter.on('warning', function (v) {
      console.error('client warning',...arguments);
  });

  b.emitter.on('error', function (v) {
    console.error('broker error:',...arguments);
  });

  c.emitter.on('error', function (v) {
    console.error('client error',...arguments);
  });

  const start = Date.now();
  const max = 17;

  let count = 0;
  let i = 0;

  async.timesLimit(10000, 30, function (n, cb) {

    const r = Math.ceil(Math.random() * 25);

    c.lock('a', {max}, (err, v) => {

      if (err) {
        return cb(err);
      }

      count++;
      console.log(i++, 'count:', count, 'max:', max);

      if (count > max) {
        throw 'count greater than max.';
      }

      setTimeout(function () {

        if (count > max) {
          throw 'count greater than max.';
        }

        count--;
        v.unlock(cb);

      }, r);

    });

  }, function (err) {

    if (err) throw err;
    console.log('all done:', Date.now() - start);
  });

});

by default max is 1. implementing the reader/writer lock is harder than I thought but I think I can figure it out in the next week or two.

from live-mutex.

@hkjeffchan try live-mutex version "0.1.1054" on NPM. You can use the semaphore feature like above, using the max:integer option.

from live-mutex.