Synchronizing access to store

Marco Lettere <m.lettere at gmail.com> schrieb am Mi., 22. Jan. 2025, 17:21: > Dear all, > > I'm wondering why the code in [1] is not working as expected. Expected > means getting back a map with 100 keys. > > Instead I get a random number of keys which denotes the fact (proved by > tracing) that the map in store is updated concurrently without > respecting the lock annotations. > > Is there a way to obtain synchronized access to key/values of the > default store? > > Thank you. > > Kind regards, > > Marco. > > [1] > > declare %updating %basex:lock('CONFIG') function local:write($k as > xs:string) { > store:put("config", map:put(store:get("config"), $k, > random:integer(10))) > }; > > let $s1 := store:put("config", map{}) > let $s2 := xquery:fork-join(for $i in (1 to 100) return function(){ > local:write("key" || $i) }) > return count(map:keys(store:get("config"))) > >

On 27/01/25 17:15, Christian Grün wrote: > Hi Marco, > > even if %basex:lock can be added to specific function calls, it will > only be considered before the full query is evaluated. As a > consequence, your current query will only be evaluated if no other > updating query with CONFIG locks is running at the same time. > > Apart from that, the store functions are not handled by our > transaction management. Instead, they will be immediately executed, > and we do our best to ensure that parallel store operations do not > result in inconsistent states (basically by treating each store update > as an atomic operation). We may need to clarify this in our documentation. > > What has been your motivation for adding entries to the store in parallel? > > Best, > Christian > > > Marco Lettere <m.lettere at gmail.com> schrieb am Mi., 22. Jan. 2025, 17:21: > > Dear all, > > I'm wondering why the code in [1] is not working as expected. > Expected > means getting back a map with 100 keys. > > Instead I get a random number of keys which denotes the fact > (proved by > tracing) that the map in store is updated concurrently without > respecting the lock annotations. > > Is there a way to obtain synchronized access to key/values of the > default store? > > Thank you. > > Kind regards, > > Marco. > > [1] > > declare %updating %basex:lock('CONFIG') function local:write($k as > xs:string) { > store:put("config", map:put(store:get("config"), $k, > random:integer(10))) > }; > > let $s1 := store:put("config", map{}) > let $s2 := xquery:fork-join(for $i in (1 to 100) return function(){ > local:write("key" || $i) }) > return count(map:keys(store:get("config"))) >

On Mon, 27 Jan 2025 at 16:41, Marco Lettere <m.lettere at gmail.com> wrote: > Thanks for checking into this Christian. > > I'm experimenting with some sort of workflow oriented programming because > our projects are getting more and more complex and we'd like to have a way > of "standardizing" some patterns and getting a bit of introspection and > observability of what happens to a complex workflow. > > We'd love to use the store as a shared memory among functions annotated as > tasks for many reasons such as keeping a global log of the workflow, > reducing dependency on function signatures or chaining together tasks > executed in subsequent continuations (aka subworkflows). > > Those are kind of things that you can benefit from when using workflow > engines. But we don't want to rely on those because they have a huge impact > in terms of provisioning and maintenance. So we wanted to check whether a > light weight solution in code is viable. > > I'm currently rewriting the code to use a DB instead of the store. But it > would have been so nice to have it without requiring an external db > server... But for example, as soon as I have two tasks running in parallel > (fork-join) and they want to add a log line to the sequence of logs that > introduces the risk of loosing data. > > Hope it clears up the intent even if a bit convoluted explanation. :-) > Ciao, > M. > > On 27/01/25 17:15, Christian Grün wrote: > > Hi Marco, > > even if %basex:lock can be added to specific function calls, it will only > be considered before the full query is evaluated. As a consequence, your > current query will only be evaluated if no other updating query with CONFIG > locks is running at the same time. > > Apart from that, the store functions are not handled by our transaction > management. Instead, they will be immediately executed, and we do our best > to ensure that parallel store operations do not result in inconsistent > states (basically by treating each store update as an atomic operation). We > may need to clarify this in our documentation. > > What has been your motivation for adding entries to the store in parallel? > > Best, > Christian > > > Marco Lettere <m.lettere at gmail.com> schrieb am Mi., 22. Jan. 2025, 17:21: > >> Dear all, >> >> I'm wondering why the code in [1] is not working as expected. Expected >> means getting back a map with 100 keys. >> >> Instead I get a random number of keys which denotes the fact (proved by >> tracing) that the map in store is updated concurrently without >> respecting the lock annotations. >> >> Is there a way to obtain synchronized access to key/values of the >> default store? >> >> Thank you. >> >> Kind regards, >> >> Marco. >> >> [1] >> >> declare %updating %basex:lock('CONFIG') function local:write($k as >> xs:string) { >> store:put("config", map:put(store:get("config"), $k, >> random:integer(10))) >> }; >> >> let $s1 := store:put("config", map{}) >> let $s2 := xquery:fork-join(for $i in (1 to 100) return function(){ >> local:write("key" || $i) }) >> return count(map:keys(store:get("config"))) >> >>

On 27/01/25 18:30, Andy Bunce wrote: > Hi Marco, > > When I try your code (with 10.7 or 11.6 ) I get > > [XPTY0004] Function is updating: %updating fn() as empty-sequence()... > > Which is what I would expect from the hint in the documentation [1] > Do you not get this error message? > > /Andy > > [1] https://docs.basex.org/main/XQuery_Functions#xquery:fork-join > > On Mon, 27 Jan 2025 at 16:41, Marco Lettere <m.lettere at gmail.com> wrote: > > Thanks for checking into this Christian. > > I'm experimenting with some sort of workflow oriented programming > because our projects are getting more and more complex and we'd > like to have a way of "standardizing" some patterns and getting a > bit of introspection and observability of what happens to a > complex workflow. > > We'd love to use the store as a shared memory among functions > annotated as tasks for many reasons such as keeping a global log > of the workflow, reducing dependency on function signatures or > chaining together tasks executed in subsequent continuations (aka > subworkflows). > > Those are kind of things that you can benefit from when using > workflow engines. But we don't want to rely on those because they > have a huge impact in terms of provisioning and maintenance. So we > wanted to check whether a light weight solution in code is viable. > > I'm currently rewriting the code to use a DB instead of the store. > But it would have been so nice to have it without requiring an > external db server... But for example, as soon as I have two tasks > running in parallel (fork-join) and they want to add a log line to > the sequence of logs that introduces the risk of loosing data. > > Hope it clears up the intent even if a bit convoluted explanation. :-) > > Ciao, > M. > > On 27/01/25 17:15, Christian Grün wrote: >> Hi Marco, >> >> even if %basex:lock can be added to specific function calls, it >> will only be considered before the full query is evaluated. As a >> consequence, your current query will only be evaluated if no >> other updating query with CONFIG locks is running at the same time. >> >> Apart from that, the store functions are not handled by our >> transaction management. Instead, they will be immediately >> executed, and we do our best to ensure that parallel store >> operations do not result in inconsistent states (basically by >> treating each store update as an atomic operation). We may need >> to clarify this in our documentation. >> >> What has been your motivation for adding entries to the store in >> parallel? >> >> Best, >> Christian >> >> >> Marco Lettere <m.lettere at gmail.com> schrieb am Mi., 22. Jan. >> 2025, 17:21: >> >> Dear all, >> >> I'm wondering why the code in [1] is not working as expected. >> Expected >> means getting back a map with 100 keys. >> >> Instead I get a random number of keys which denotes the fact >> (proved by >> tracing) that the map in store is updated concurrently without >> respecting the lock annotations. >> >> Is there a way to obtain synchronized access to key/values of >> the >> default store? >> >> Thank you. >> >> Kind regards, >> >> Marco. >> >> [1] >> >> declare %updating %basex:lock('CONFIG') function >> local:write($k as >> xs:string) { >> store:put("config", map:put(store:get("config"), $k, >> random:integer(10))) >> }; >> >> let $s1 := store:put("config", map{}) >> let $s2 := xquery:fork-join(for $i in (1 to 100) return >> function(){ >> local:write("key" || $i) }) >> return count(map:keys(store:get("config"))) >>

On Mon, 27 Jan 2025 at 17:37, Marco Lettere <m.lettere at gmail.com> wrote: > Hi Andy, > > sorry for not specifying but we usually set the following property in the > .basex file in order to being able to blend updating and non updating > functions. > > # Local Options > MIXUPDATES = true > > M. > On 27/01/25 18:30, Andy Bunce wrote: > > Hi Marco, > > When I try your code (with 10.7 or 11.6 ) I get > > [XPTY0004] Function is updating: %updating fn() as empty-sequence()... > > Which is what I would expect from the hint in the documentation [1] > Do you not get this error message? > > /Andy > > [1] https://docs.basex.org/main/XQuery_Functions#xquery:fork-join > > On Mon, 27 Jan 2025 at 16:41, Marco Lettere <m.lettere at gmail.com> wrote: > >> Thanks for checking into this Christian. >> >> I'm experimenting with some sort of workflow oriented programming because >> our projects are getting more and more complex and we'd like to have a way >> of "standardizing" some patterns and getting a bit of introspection and >> observability of what happens to a complex workflow. >> >> We'd love to use the store as a shared memory among functions annotated >> as tasks for many reasons such as keeping a global log of the workflow, >> reducing dependency on function signatures or chaining together tasks >> executed in subsequent continuations (aka subworkflows). >> >> Those are kind of things that you can benefit from when using workflow >> engines. But we don't want to rely on those because they have a huge impact >> in terms of provisioning and maintenance. So we wanted to check whether a >> light weight solution in code is viable. >> >> I'm currently rewriting the code to use a DB instead of the store. But it >> would have been so nice to have it without requiring an external db >> server... But for example, as soon as I have two tasks running in parallel >> (fork-join) and they want to add a log line to the sequence of logs that >> introduces the risk of loosing data. >> >> Hope it clears up the intent even if a bit convoluted explanation. :-) >> Ciao, >> M. >> >> On 27/01/25 17:15, Christian Grün wrote: >> >> Hi Marco, >> >> even if %basex:lock can be added to specific function calls, it will only >> be considered before the full query is evaluated. As a consequence, your >> current query will only be evaluated if no other updating query with CONFIG >> locks is running at the same time. >> >> Apart from that, the store functions are not handled by our transaction >> management. Instead, they will be immediately executed, and we do our best >> to ensure that parallel store operations do not result in inconsistent >> states (basically by treating each store update as an atomic operation). We >> may need to clarify this in our documentation. >> >> What has been your motivation for adding entries to the store in parallel? >> >> Best, >> Christian >> >> >> Marco Lettere <m.lettere at gmail.com> schrieb am Mi., 22. Jan. 2025, 17:21: >> >>> Dear all, >>> >>> I'm wondering why the code in [1] is not working as expected. Expected >>> means getting back a map with 100 keys. >>> >>> Instead I get a random number of keys which denotes the fact (proved by >>> tracing) that the map in store is updated concurrently without >>> respecting the lock annotations. >>> >>> Is there a way to obtain synchronized access to key/values of the >>> default store? >>> >>> Thank you. >>> >>> Kind regards, >>> >>> Marco. >>> >>> [1] >>> >>> declare %updating %basex:lock('CONFIG') function local:write($k as >>> xs:string) { >>> store:put("config", map:put(store:get("config"), $k, >>> random:integer(10))) >>> }; >>> >>> let $s1 := store:put("config", map{}) >>> let $s2 := xquery:fork-join(for $i in (1 to 100) return function(){ >>> local:write("key" || $i) }) >>> return count(map:keys(store:get("config"))) >>> >>>

>so if you call store:get, store:put or store:write in the first process,

On Tue, 28 Jan 2025 at 14:09, Marco Lettere <m.lettere at gmail.com> wrote: > Ok, thanks for the clarification. > > M. > On 28/01/25 15:08, Christian Grün wrote: > > Sorry Christian, do you mean *not* synchronized? >> > With »synchronized«, I meant to refer to a lower level: You will not end > up with a corrupt key/value store or with I/O conflicts when accessing and > updating the store via multiple threads. However, as you have already > observed, multiple operations are not executed in a well-defined order, so > if you call store:get, store:put or store:write in the first process, a > second process will not wait until the store operations are completed. > >

On 07/02/25 16:45, Andy Bunce wrote: > >so if you call store:get, store:put or store:write in the first process, a > second process will not wait until the store operations are completed. > > In non XQuery contexts a semaphore [1] might be used to ensure that my > other threads don't get between a get and put. > In the spirit of blurring the XQuery Java boundaries I tried [2]. It > seems to work. Is it dangerous? > > [1] > https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Semaphore.html > > [2] > declare namespace Semphore ="java:java.util.concurrent.Semaphore"; > > declare function local:config-update($k as xs:string,$v as item(),$sem) { > Semphore:acquire($sem), > try{ > let $u:=store:get("config")=> map:put($k, $v) > return store:put("config",$u) > }catch * { > trace("Errrr",$err:description) > } > ,Semphore:release($sem) > }; > > let $sem:=Semphore:new(1,true()) > > let $s1 := store:put("config", map{}) > let $s2 := xquery:fork-join( > for $i in (1 to 100) > return function(){ > let $r:=(prof:sleep(10),$i) > return local:config-update( string($i),$r,$sem) } > ) > > return count(map:keys(store:get("config"))) > > > On Tue, 28 Jan 2025 at 14:09, Marco Lettere <m.lettere at gmail.com> wrote: > > Ok, thanks for the clarification. > > M. > > On 28/01/25 15:08, Christian Grün wrote: >> >> Sorry Christian, do you mean *not* synchronized? >> >> With »synchronized«, I meant to refer to a lower level: You will >> not end up with a corrupt key/value store or with I/O conflicts >> when accessing and updating the store via multiple >> threads. However, as you have already observed, multiple >> operations are not executed in a well-defined order, so if you >> call store:get, store:put or store:write in the first process, a >> second process will not wait until the store operations are >> completed. >>

On Fri, 7 Feb 2025 at 18:31, Marco Lettere <m.lettere at gmail.com> wrote: > Oh, wow. Looks great Andy. Thanks for suggestion. > > Wonder to know what's Christian's opinion on this. > > M. > On 07/02/25 16:45, Andy Bunce wrote: > > >so if you call store:get, store:put or store:write in the first process, > a second process will not wait until the store operations are completed. > > In non XQuery contexts a semaphore [1] might be used to ensure that my > other threads don't get between a get and put. > In the spirit of blurring the XQuery Java boundaries I tried [2]. It seems > to work. Is it dangerous? > > [1] > https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Semaphore.html > > [2] > declare namespace Semphore ="java:java.util.concurrent.Semaphore"; > > declare function local:config-update($k as xs:string,$v as item(),$sem) { > Semphore:acquire($sem), > try{ > let $u:=store:get("config")=> map:put($k, $v) > return store:put("config",$u) > }catch * { > trace("Errrr",$err:description) > } > ,Semphore:release($sem) > }; > > let $sem:=Semphore:new(1,true()) > > let $s1 := store:put("config", map{}) > let $s2 := xquery:fork-join( > for $i in (1 to 100) > return function(){ > let $r:=(prof:sleep(10),$i) > return local:config-update( string($i),$r,$sem) } > ) > > return count(map:keys(store:get("config"))) > > > On Tue, 28 Jan 2025 at 14:09, Marco Lettere <m.lettere at gmail.com> wrote: > >> Ok, thanks for the clarification. >> >> M. >> On 28/01/25 15:08, Christian Grün wrote: >> >> Sorry Christian, do you mean *not* synchronized? >>> >> With »synchronized«, I meant to refer to a lower level: You will not end >> up with a corrupt key/value store or with I/O conflicts when accessing and >> updating the store via multiple threads. However, as you have already >> observed, multiple operations are not executed in a well-defined order, so >> if you call store:get, store:put or store:write in the first process, a >> second process will not wait until the store operations are completed. >> >>

Il sab 8 feb 2025, 11:55 Andy Bunce <bunce.andy at gmail.com> ha scritto: > One problem with this might be that: > When using the fork-join function it is easy to ensure all the threads > have a reference to the *same* semaphore. > If these were arbitrary BaseX "jobs" it is not clear how this could be > done without explicit support from the BaseX runtime. > Perhaps it could be done with a new annotation > %basex:semaphore ("my-semaphore") > that could be applied to functions. > > /Andy > > On Fri, 7 Feb 2025 at 18:31, Marco Lettere <m.lettere at gmail.com> wrote: > >> Oh, wow. Looks great Andy. Thanks for suggestion. >> >> Wonder to know what's Christian's opinion on this. >> >> M. >> On 07/02/25 16:45, Andy Bunce wrote: >> >> >so if you call store:get, store:put or store:write in the first process, >> a second process will not wait until the store operations are completed. >> >> In non XQuery contexts a semaphore [1] might be used to ensure that my >> other threads don't get between a get and put. >> In the spirit of blurring the XQuery Java boundaries I tried [2]. It >> seems to work. Is it dangerous? >> >> [1] >> https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Semaphore.html >> >> [2] >> declare namespace Semphore ="java:java.util.concurrent.Semaphore"; >> >> declare function local:config-update($k as xs:string,$v as item(),$sem) { >> Semphore:acquire($sem), >> try{ >> let $u:=store:get("config")=> map:put($k, $v) >> return store:put("config",$u) >> }catch * { >> trace("Errrr",$err:description) >> } >> ,Semphore:release($sem) >> }; >> >> let $sem:=Semphore:new(1,true()) >> >> let $s1 := store:put("config", map{}) >> let $s2 := xquery:fork-join( >> for $i in (1 to 100) >> return function(){ >> let $r:=(prof:sleep(10),$i) >> return local:config-update( string($i),$r,$sem) } >> ) >> >> return count(map:keys(store:get("config"))) >> >> >> On Tue, 28 Jan 2025 at 14:09, Marco Lettere <m.lettere at gmail.com> wrote: >> >>> Ok, thanks for the clarification. >>> >>> M. >>> On 28/01/25 15:08, Christian Grün wrote: >>> >>> Sorry Christian, do you mean *not* synchronized? >>>> >>> With »synchronized«, I meant to refer to a lower level: You will not end >>> up with a corrupt key/value store or with I/O conflicts when accessing and >>> updating the store via multiple threads. However, as you have already >>> observed, multiple operations are not executed in a well-defined order, so >>> if you call store:get, store:put or store:write in the first process, a >>> second process will not wait until the store operations are completed. >>> >>>

On Sat, Feb 8, 2025 at 12:07 PM Marco Lettere <m.lettere at gmail.com> wrote: > For us it can be solved with custom annotations or even constant strings > because it all comes down to the one workflow engine ... > > Il sab 8 feb 2025, 11:55 Andy Bunce <bunce.andy at gmail.com> ha scritto: > >> One problem with this might be that: >> When using the fork-join function it is easy to ensure all the threads >> have a reference to the *same* semaphore. >> If these were arbitrary BaseX "jobs" it is not clear how this could be >> done without explicit support from the BaseX runtime. >> Perhaps it could be done with a new annotation >> %basex:semaphore ("my-semaphore") >> that could be applied to functions. >> >> /Andy >> >> On Fri, 7 Feb 2025 at 18:31, Marco Lettere <m.lettere at gmail.com> wrote: >> >>> Oh, wow. Looks great Andy. Thanks for suggestion. >>> >>> Wonder to know what's Christian's opinion on this. >>> >>> M. >>> On 07/02/25 16:45, Andy Bunce wrote: >>> >>> >so if you call store:get, store:put or store:write in the first >>> process, a second process will not wait until the store operations are >>> completed. >>> >>> In non XQuery contexts a semaphore [1] might be used to ensure that my >>> other threads don't get between a get and put. >>> In the spirit of blurring the XQuery Java boundaries I tried [2]. It >>> seems to work. Is it dangerous? >>> >>> [1] >>> https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Semaphore.html >>> >>> [2] >>> declare namespace Semphore ="java:java.util.concurrent.Semaphore"; >>> >>> declare function local:config-update($k as xs:string,$v as item(),$sem) >>> { >>> Semphore:acquire($sem), >>> try{ >>> let $u:=store:get("config")=> map:put($k, $v) >>> return store:put("config",$u) >>> }catch * { >>> trace("Errrr",$err:description) >>> } >>> ,Semphore:release($sem) >>> }; >>> >>> let $sem:=Semphore:new(1,true()) >>> >>> let $s1 := store:put("config", map{}) >>> let $s2 := xquery:fork-join( >>> for $i in (1 to 100) >>> return function(){ >>> let $r:=(prof:sleep(10),$i) >>> return local:config-update( string($i),$r,$sem) } >>> ) >>> >>> return count(map:keys(store:get("config"))) >>> >>> >>> On Tue, 28 Jan 2025 at 14:09, Marco Lettere <m.lettere at gmail.com> wrote: >>> >>>> Ok, thanks for the clarification. >>>> >>>> M. >>>> On 28/01/25 15:08, Christian Grün wrote: >>>> >>>> Sorry Christian, do you mean *not* synchronized? >>>>> >>>> With »synchronized«, I meant to refer to a lower level: You will not >>>> end up with a corrupt key/value store or with I/O conflicts when accessing >>>> and updating the store via multiple threads. However, as you have already >>>> observed, multiple operations are not executed in a well-defined order, so >>>> if you call store:get, store:put or store:write in the first process, a >>>> second process will not wait until the store operations are completed. >>>> >>>>

On 22/02/25 11:38, Christian Grün wrote: > Hi there, > > The semaphore code looks fine. As Andy noted, things get much more > complicated when the code needs to run across multiple concurrent > jobs: We might easily run into deadlocks. > > Thus, if it turns out that concurrency gets a bigger issue, we would > probably embed key/value store updates into our Pending Update List > concept, even if it would make it less flexible to use. > > Cheers, > Christian > > > On Sat, Feb 8, 2025 at 12:07 PM Marco Lettere <m.lettere at gmail.com> wrote: > > For us it can be solved with custom annotations or even constant > strings because it all comes down to the one workflow engine ... > > > Il sab 8 feb 2025, 11:55 Andy Bunce <bunce.andy at gmail.com> ha scritto: > > One problem with this might be that: > When using the fork-join function it is easy to ensure all the > threads have a reference to the *same* semaphore. > If these were arbitrary BaseX "jobs" it is not clear how this > could be done without explicit support from the BaseX runtime. > Perhaps it could be done with a new annotation > %basex:semaphore ("my-semaphore") > that could be applied to functions. > > /Andy > > On Fri, 7 Feb 2025 at 18:31, Marco Lettere > <m.lettere at gmail.com> wrote: > > Oh, wow. Looks great Andy. Thanks for suggestion. > > Wonder to know what's Christian's opinion on this. > > M. > > On 07/02/25 16:45, Andy Bunce wrote: >> >so if you call store:get, store:put or store:write in the first >> process, a second process will not wait until the store >> operations are completed. >> >> In non XQuery contexts a semaphore [1] might be used to >> ensure that my other threads don't get between a get and >> put. >> In the spirit of blurring the XQuery Java boundaries I >> tried [2]. It seems to work. Is it dangerous? >> >> [1] >> https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Semaphore.html >> >> [2] >> declare namespace Semphore >> ="java:java.util.concurrent.Semaphore"; >> >> declare function local:config-update($k as xs:string,$v >> as item(),$sem) { >> Semphore:acquire($sem), >> try{ >> let $u:=store:get("config")=> map:put($k, $v) >> return store:put("config",$u) >> }catch * { >> trace("Errrr",$err:description) >> } >> ,Semphore:release($sem) >> }; >> >> let $sem:=Semphore:new(1,true()) >> >> let $s1 := store:put("config", map{}) >> let $s2 := xquery:fork-join( >> for $i in (1 to 100) >> return function(){ >> let $r:=(prof:sleep(10),$i) >> return local:config-update( string($i),$r,$sem) } >> ) >> >> return count(map:keys(store:get("config"))) >> >> >> On Tue, 28 Jan 2025 at 14:09, Marco Lettere >> <m.lettere at gmail.com> wrote: >> >> Ok, thanks for the clarification. >> >> M. >> >> On 28/01/25 15:08, Christian Grün wrote: >>> >>> Sorry Christian, do you mean *not* synchronized? >>> >>> With »synchronized«, I meant to refer to a lower >>> level: You will not end up with a corrupt key/value >>> store or with I/O conflicts when accessing and >>> updating the store via multiple threads. However, as >>> you have already observed, multiple operations are >>> not executed in a well-defined order, so if you call >>> store:get, store:put or store:write in the first >>> process, a second process will not wait until the >>> store operations are completed. >>>