Browse files

[fix] runtime: Despite atomic directives sections could be scheduled …

…cause of Hashtbl.make which implemented by a non-cps and second order bypass, Credit:Bug reported by Alok Menghrajani: http://forum.opalang.org/1_318
  • Loading branch information...
1 parent a46e261 commit fbc4bf05efc93756c880843f56e3861d31d351aa @BourgerieQuentin BourgerieQuentin committed Oct 11, 2012
Showing with 77 additions and 42 deletions.
  1. +14 −12 lib/stdlib/core/rpc/core/client_event.opa
  2. +63 −30 lib/stdlib/core/rpc/core/ping_register.opa
View
26 lib/stdlib/core/rpc/core/client_event.opa
@@ -45,12 +45,14 @@ ClientEvent = {{
@private
error = Log.error("CLIENTEVENT", _)
+ @private ClientTbl = PingRegister.ClientTbl
+
/**
* The table which store the state of client connection
*/
@private
- state_tbl : Hashtbl.t(ThreadContext.client, PingRegister.state) =
- PingRegister.Utils.make_tctbl()
+ state_tbl : ClientTbl.t(PingRegister.state) =
+ ClientTbl.create()
/**
* Default delay before raise the disconnect event
@@ -74,13 +76,13 @@ ClientEvent = {{
if kind == event then Scheduler.push(-> cb(client))
, binds)
do aux(binds)
- match Hashtbl.try_find(state_tbl, ThreadContext.Client.fake)
+ match ClientTbl.try_find(state_tbl, ThreadContext.Client.fake)
| {none} -> void
| {some = state} -> aux(Hashtbl.bindings(state.callbacks))
/* The fake client is used for to store default handlers */
@private
- _add_fake = Hashtbl.add(state_tbl, ThreadContext.Client.fake, {
+ _add_fake = ClientTbl.add(state_tbl, ThreadContext.Client.fake, {
ina_delay = Reference.create(-1)
ina_key = Reference.create({none})
dis_delay = Reference.create(-1)
@@ -131,7 +133,7 @@ ClientEvent = {{
client = match client with
| {none} -> ThreadContext.Client.fake
| {some = client} -> client
- match Hashtbl.try_find(state_tbl, client)
+ match ClientTbl.try_find(state_tbl, client)
| {none} ->
do error("Event({client}) No register callback event, client is not present")
@fail
@@ -172,7 +174,7 @@ ClientEvent = {{
* @param binding_id The id of the binding event/callback to remove.
*/
remove_event(key:ClientEvent.key) =
- @atomic(match Hashtbl.try_find(state_tbl, key.client) with
+ @atomic(match ClientTbl.try_find(state_tbl, key.client) with
| {none} -> void
| {some = state} ->
Hashtbl.remove(state.callbacks, key.ckey))
@@ -216,7 +218,7 @@ ClientEvent = {{
match ctx with
| {none} -> Reference.set(inactive_delay, time)
| {some=client} ->
- match Hashtbl.try_find(state_tbl, client) with
+ match ClientTbl.try_find(state_tbl, client) with
| {none} -> void
| {some = state} -> Reference.set(state.ina_delay, time ? 0)
@@ -254,10 +256,10 @@ ClientEvent = {{
#<Ifstatic:MLSTATE_PING_DEBUG>
do debug("Event({client}) Remove client state")
#<End>
- match @atomic(match Hashtbl.try_find(state_tbl, client) with
+ match @atomic(match ClientTbl.try_find(state_tbl, client) with
| {none} -> {}
| {some = _} as e ->
- do Hashtbl.remove(state_tbl, client)
+ do ClientTbl.remove(state_tbl, client)
e)
with
| {} -> void
@@ -280,7 +282,7 @@ ClientEvent = {{
do debug("Event({client}) Touch client state")
#<End>
~{delay ref iref idelay} =
- @atomic(match Hashtbl.try_find(state_tbl, client) with
+ @atomic(match ClientTbl.try_find(state_tbl, client) with
| {some = state} ->
aux(ref) =
match Reference.get(ref) with
@@ -302,7 +304,7 @@ ClientEvent = {{
dis_key = Reference.create({none})
ina_key = Reference.create({none})
}
- do Hashtbl.add(state_tbl, client, state)
+ do ClientTbl.add(state_tbl, client, state)
{delay = Reference.get(state.dis_delay) ref=state.dis_key
idelay = Reference.get(state.ina_delay) iref = state.ina_key}
)
@@ -316,7 +318,7 @@ ClientEvent = {{
do if active && idelay > 0 then
k = Scheduler.asleep(idelay,
-> Option.iter(s -> raise_event(client, {inactive}, Hashtbl.bindings(s.callbacks))
- , Hashtbl.try_find(state_tbl, client))
+ , ClientTbl.try_find(state_tbl, client))
)
aux(iref, k)
aux(ref, Scheduler.asleep(delay, -> remove(client)))
View
93 lib/stdlib/core/rpc/core/ping_register.opa
@@ -22,6 +22,9 @@ type PingRegister.entry =
@private
type PingRegister.pang_result = {nb:int result:string}
+@abstract
+type ClientTbl.t('a) = Hashtbl.t(string, 'a)
+
@server_private
PingRegister = {{
@@ -32,14 +35,25 @@ PingRegister = {{
error = Log.error("PING", _)
@package
- Utils = {{
+ ClientTbl = {{
@private
hash_client(~{client page}:ThreadContext.client) =
- "{client}_{page}"
+ @atomic(String.of_int(page) ^ "_" ^ client)
+
+ create() : ClientTbl.t('a) = Hashtbl.create(1024)
+
+ add(tbl, client, value) =
+ Hashtbl.add(tbl, hash_client(client), value)
+
+ replace(tbl, client, value) =
+ Hashtbl.replace(tbl, hash_client(client), value)
- make_tctbl(): Hashtbl.t(ThreadContext.client, 'a) =
- Hashtbl.make(hash_client, (_, _ -> true), 1024)
+ remove(tbl, client) =
+ Hashtbl.remove(tbl, hash_client(client))
+
+ try_find(tbl, client) =
+ Hashtbl.try_find(tbl, hash_client(client))
}}
@@ -55,12 +69,12 @@ PingRegister = {{
"{client}_{page}"
@private
- entries : Hashtbl.t(ThreadContext.client, PingRegister.entry) =
- Utils.make_tctbl()
+ entries : ClientTbl.t(PingRegister.entry) =
+ ClientTbl.create()
@private
- pangtbl : Hashtbl.t(ThreadContext.client, list(PingRegister.pang_result)) =
- Utils.make_tctbl()
+ pangtbl : ClientTbl.t(list(PingRegister.pang_result)) =
+ ClientTbl.create()
@private
ping_delay = 30 * 1000
@@ -70,32 +84,31 @@ PingRegister = {{
*/
@private
add(client, entry) =
- //do @assert(not(Hashtbl.mem(entries, client)))
- Hashtbl.add(entries, client, entry)
+ ClientTbl.add(entries, client, entry)
/**
* As add but overwrite the previous association if already exists.
*/
@private
replace(client, entry) =
- Hashtbl.replace(entries, client, entry)
+ ClientTbl.replace(entries, client, entry)
/**
* Remove the entry association to the [client]
*/
@private
remove(client) =
- Hashtbl.remove(entries, client)
+ ClientTbl.remove(entries, client)
/**
* Add the [nb]th pang [result] to the pang table.
*/
@private
add_pang(client, nb, result) =
@atomic(
- match Hashtbl.try_find(pangtbl, client) with
- | {none} -> Hashtbl.add(pangtbl, client, [~{result nb}])
- | {some = pangs} -> Hashtbl.replace(pangtbl, client, [~{result nb} | pangs])
+ match ClientTbl.try_find(pangtbl, client) with
+ | {none} -> ClientTbl.add(pangtbl, client, [~{result nb}])
+ | {some = pangs} -> ClientTbl.replace(pangtbl, client, [~{result nb} | pangs])
)
/**
@@ -122,7 +135,7 @@ PingRegister = {{
do debug("PONG({pnb}, {client}) sending")
#<End>
match @atomic(
- match Hashtbl.try_find(entries, client) with
+ match ClientTbl.try_find(entries, client) with
| {some = ~{ajax nb key=_}} ->
do remove(client)
if pnb == nb then {~ajax pong}
@@ -155,7 +168,7 @@ PingRegister = {{
#<End>
Scheduler.asleep(ping_delay, -> pong(client, nb))
match @atomic(
- match Hashtbl.try_find(entries, client) with
+ match ClientTbl.try_find(entries, client) with
| {none} ->
do add(client, {ajax=winfo ~nb key=apong})
{}
@@ -175,12 +188,24 @@ PingRegister = {{
do Scheduler.abort(apong)
{error already=onb}
) with
- | {} -> void
+ | {} ->
+ #<Ifstatic:MLSTATE_PING_DEBUG>
+ do debug("PING({client}) nothing to do, waiting messages")
+ #<End>
+ void
| {error ~already} ->
do error("PING({nb}, {client}) not registered PING({already}) already present")
send_response(winfo, {break})
- | {~winfo break} -> send_response(winfo, {break})
- | {msgs=_} as e -> send_response(winfo, e)
+ | {~winfo break} ->
+ #<Ifstatic:MLSTATE_PING_DEBUG>
+ do debug("SEND({client}) break the ping loop")
+ #<End>
+ send_response(winfo, {break})
+ | {msgs=_} as e ->
+ #<Ifstatic:MLSTATE_PING_DEBUG>
+ do debug("SEND({client}) flush pending messages {e}")
+ #<End>
+ send_response(winfo, e)
/**
* As [iping] plus handles pang mecanism.
@@ -190,18 +215,22 @@ PingRegister = {{
do debug("PING({nb}, {client}) is received")
#<End>
match @atomic(
- match Hashtbl.try_find(pangtbl, client) with
+ match ClientTbl.try_find(pangtbl, client) with
| {none}
| {some = []} -> {}
| {some = [pang|pangs]} ->
- do Hashtbl.replace(pangtbl, client, pangs)
+ do ClientTbl.replace(pangtbl, client, pangs)
do match pangs with
- | [] -> Hashtbl.remove(pangtbl, client)
+ | [] -> ClientTbl.remove(pangtbl, client)
| _ -> void
~{winfo pang}
) with
| {} -> iping(crush, client, nb, winfo)
- | ~{winfo pang} -> send_response(winfo, @opensums(pang))
+ | ~{winfo pang} ->
+ #<Ifstatic:MLSTATE_PING_DEBUG>
+ do debug("PANG({pang.nb}, {client}) return waiting pang")
+ #<End>
+ send_response(winfo, @opensums(pang))
/**
* Handles a pang request, it's like a synchronous ping.
@@ -220,7 +249,7 @@ PingRegister = {{
do debug("PANG({nb}, {client}) return ")
#<End>
match @atomic(
- match Hashtbl.try_find(entries, client) with
+ match ClientTbl.try_find(entries, client) with
| {some = ~{ajax key ...}} ->
do remove(client)
do Scheduler.abort(key)
@@ -230,24 +259,28 @@ PingRegister = {{
do add_pang(client, nb, result)
{}
) with
- | ~{ajax} -> send_response(ajax, ~{result nb})
+ | ~{ajax} ->
+ #<Ifstatic:MLSTATE_PING_DEBUG>
+ do debug("PANG({nb}, {client}) direct return")
+ #<End>
+ send_response(ajax, ~{result nb})
| {} -> void
send(client, msg:PingRegister.msg) =
#<Ifstatic:MLSTATE_PING_DEBUG>
do debug("SEND({client}) message {msg}")
#<End>
match @atomic(
- match Hashtbl.try_find(entries, client) with
+ match ClientTbl.try_find(entries, client) with
| {some = ~{ajax key ...}} ->
- do Hashtbl.remove(entries, client)
+ do remove(client)
do Scheduler.abort(key)
~{ajax}
| {some = ~{msgs}} ->
- do Hashtbl.replace(entries, client, {msgs = [msg | msgs]})
+ do replace(client, {msgs = [msg | msgs]})
{}
| {none} ->
- do Hashtbl.add(entries, client, {msgs = [msg]})
+ do add(client, {msgs = [msg]})
{}
) with
| ~{ajax} ->

0 comments on commit fbc4bf0

Please sign in to comment.