Merge branch 'async_fd'
This commit is contained in:
@@ -820,6 +820,25 @@ function create_actor(desc) {
|
||||
|
||||
var $_ = {}
|
||||
|
||||
// Forward-declare actor system variables so closures in $_ can capture them.
|
||||
// Initialized here; values used at runtime when fully set up.
|
||||
var time = null
|
||||
var enet = null
|
||||
var HEADER = {}
|
||||
var underlings = {}
|
||||
var overling = null
|
||||
var root = null
|
||||
var receive_fn = null
|
||||
var greeters = {}
|
||||
var message_queue = []
|
||||
var couplings = {}
|
||||
var peers = {}
|
||||
var id_address = {}
|
||||
var peer_queue = {}
|
||||
var portal = null
|
||||
var portal_fn = null
|
||||
var replies = {}
|
||||
|
||||
use_cache['core/json'] = json
|
||||
|
||||
// Runtime env: passed to package modules via shop's inject_env.
|
||||
@@ -859,6 +878,205 @@ core_extras.parallel = parallel
|
||||
core_extras.race = race
|
||||
core_extras.sequence = sequence
|
||||
|
||||
// Set actor identity before shop loads so $self is available
|
||||
if (!_cell.args.id) _cell.id = guid()
|
||||
else _cell.id = _cell.args.id
|
||||
|
||||
$_.self = create_actor({id: _cell.id})
|
||||
|
||||
overling = _cell.args.overling_id ? create_actor({id: _cell.args.overling_id}) : null
|
||||
$_.overling = overling
|
||||
|
||||
root = _cell.args.root_id ? create_actor({id: _cell.args.root_id}) : null
|
||||
if (root == null) root = $_.self
|
||||
|
||||
// Define all actor intrinsic functions ($clock, $delay, etc.) before shop loads.
|
||||
// Closures here capture module-level variables by reference; those variables
|
||||
// are fully initialized before these functions are ever called at runtime.
|
||||
|
||||
$_.clock = function(fn) {
|
||||
actor_mod.clock(_ => {
|
||||
fn(time.number())
|
||||
send_messages()
|
||||
})
|
||||
}
|
||||
|
||||
$_.delay = function delay(fn, seconds) {
|
||||
var _seconds = seconds == null ? 0 : seconds
|
||||
function delay_turn() {
|
||||
fn()
|
||||
send_messages()
|
||||
}
|
||||
var id = actor_mod.delay(delay_turn, _seconds)
|
||||
return function() { actor_mod.removetimer(id) }
|
||||
}
|
||||
|
||||
$_.stop = function stop(actor) {
|
||||
if (!actor) {
|
||||
need_stop = true
|
||||
return
|
||||
}
|
||||
if (!is_actor(actor)) {
|
||||
log.error('Can only call stop on an actor.')
|
||||
disrupt
|
||||
}
|
||||
if (is_null(underlings[actor[ACTORDATA].id])) {
|
||||
log.error('Can only call stop on an underling or self.')
|
||||
disrupt
|
||||
}
|
||||
sys_msg(actor, {kind:"stop"})
|
||||
}
|
||||
|
||||
$_.start = function start(cb, program) {
|
||||
if (!program) return
|
||||
var id = guid()
|
||||
var oid = $_.self[ACTORDATA].id
|
||||
var startup = {
|
||||
id,
|
||||
overling_id: oid,
|
||||
root_id: root ? root[ACTORDATA].id : null,
|
||||
program,
|
||||
native_mode: native_mode,
|
||||
no_warn: _no_warn,
|
||||
}
|
||||
greeters[id] = cb
|
||||
push(message_queue, { startup })
|
||||
}
|
||||
|
||||
$_.receiver = function receiver(fn) {
|
||||
receive_fn = fn
|
||||
}
|
||||
|
||||
$_.unneeded = function unneeded(fn, seconds) {
|
||||
actor_mod.unneeded(fn, seconds)
|
||||
}
|
||||
|
||||
$_.couple = function couple(actor) {
|
||||
if (actor == $_.self) return
|
||||
couplings[actor[ACTORDATA].id] = true
|
||||
sys_msg(actor, {kind:'couple', from_id: _cell.id})
|
||||
log.system(`coupled to ${actor[ACTORDATA].id}`)
|
||||
}
|
||||
|
||||
$_.contact = function(callback, record) {
|
||||
send(create_actor(record), record, callback)
|
||||
}
|
||||
|
||||
$_.portal = function(fn, port) {
|
||||
if (portal) {
|
||||
log.error(`Already started a portal listening on ${portal.port()}`)
|
||||
disrupt
|
||||
}
|
||||
if (!port) {
|
||||
log.error("Requires a valid port.")
|
||||
disrupt
|
||||
}
|
||||
log.system(`starting a portal on port ${port}`)
|
||||
portal = enet.create_host({address: "any", port})
|
||||
portal_fn = fn
|
||||
enet_check()
|
||||
}
|
||||
|
||||
$_.connection = function(callback, actor, config) {
|
||||
var peer = peers[actor[ACTORDATA].id]
|
||||
if (peer) {
|
||||
callback(peer_connection(peer))
|
||||
return
|
||||
}
|
||||
if (actor_mod.mailbox_exist(actor[ACTORDATA].id)) {
|
||||
callback({type:"local"})
|
||||
return
|
||||
}
|
||||
callback()
|
||||
}
|
||||
|
||||
$_.time_limit = function(requestor, seconds) {
|
||||
if (!pronto.is_requestor(requestor)) {
|
||||
log.error('time_limit: first argument must be a requestor')
|
||||
disrupt
|
||||
}
|
||||
if (!is_number(seconds) || seconds <= 0) {
|
||||
log.error('time_limit: seconds must be a positive number')
|
||||
disrupt
|
||||
}
|
||||
|
||||
return function time_limit_requestor(callback, value) {
|
||||
pronto.check_callback(callback, 'time_limit')
|
||||
var finished = false
|
||||
var requestor_cancel = null
|
||||
var timer_cancel = null
|
||||
|
||||
function cancel(reason) {
|
||||
if (finished) return
|
||||
finished = true
|
||||
if (timer_cancel) {
|
||||
timer_cancel()
|
||||
timer_cancel = null
|
||||
}
|
||||
if (requestor_cancel) {
|
||||
requestor_cancel(reason)
|
||||
requestor_cancel = null
|
||||
}
|
||||
}
|
||||
|
||||
function safe_cancel_requestor(reason) {
|
||||
if (requestor_cancel) {
|
||||
requestor_cancel(reason)
|
||||
requestor_cancel = null
|
||||
}
|
||||
}
|
||||
|
||||
timer_cancel = $_.delay(function() {
|
||||
if (finished) return
|
||||
def reason = {
|
||||
factory: time_limit_requestor,
|
||||
excuse: 'Timeout.',
|
||||
evidence: seconds,
|
||||
message: 'Timeout. ' + text(seconds)
|
||||
}
|
||||
safe_cancel_requestor(reason)
|
||||
finished = true
|
||||
callback(null, reason)
|
||||
}, seconds)
|
||||
|
||||
function do_request() {
|
||||
requestor_cancel = requestor(function(val, reason) {
|
||||
if (finished) return
|
||||
finished = true
|
||||
if (timer_cancel) {
|
||||
timer_cancel()
|
||||
timer_cancel = null
|
||||
}
|
||||
callback(val, reason)
|
||||
}, value)
|
||||
} disruption {
|
||||
cancel('requestor failed')
|
||||
callback(null, 'requestor failed')
|
||||
}
|
||||
do_request()
|
||||
|
||||
return function(reason) {
|
||||
safe_cancel_requestor(reason)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Make actor intrinsics available to core modules loaded via use_core
|
||||
core_extras['$self'] = $_.self
|
||||
core_extras['$overling'] = $_.overling
|
||||
core_extras['$clock'] = $_.clock
|
||||
core_extras['$delay'] = $_.delay
|
||||
core_extras['$start'] = $_.start
|
||||
core_extras['$stop'] = $_.stop
|
||||
core_extras['$receiver'] = $_.receiver
|
||||
core_extras['$contact'] = $_.contact
|
||||
core_extras['$portal'] = $_.portal
|
||||
core_extras['$time_limit'] = $_.time_limit
|
||||
core_extras['$couple'] = $_.couple
|
||||
core_extras['$unneeded'] = $_.unneeded
|
||||
core_extras['$connection'] = $_.connection
|
||||
core_extras['$fd'] = fd
|
||||
|
||||
// NOW load shop -- it receives all of the above via env
|
||||
var shop = use_core('internal/shop')
|
||||
use_core('build')
|
||||
@@ -878,7 +1096,7 @@ _summary_resolver = function(path, ctx) {
|
||||
return summary_fn()
|
||||
}
|
||||
|
||||
var time = use_core('time')
|
||||
time = use_core('time')
|
||||
var toml = use_core('toml')
|
||||
|
||||
// --- Logging system (full version) ---
|
||||
@@ -1097,73 +1315,6 @@ runtime_env.sequence = sequence
|
||||
// Make runtime functions available to modules loaded via use_core
|
||||
arrfor(array(runtime_env), function(k) { core_extras[k] = runtime_env[k] })
|
||||
|
||||
$_.time_limit = function(requestor, seconds)
|
||||
{
|
||||
if (!pronto.is_requestor(requestor)) {
|
||||
log.error('time_limit: first argument must be a requestor')
|
||||
disrupt
|
||||
}
|
||||
if (!is_number(seconds) || seconds <= 0) {
|
||||
log.error('time_limit: seconds must be a positive number')
|
||||
disrupt
|
||||
}
|
||||
|
||||
return function time_limit_requestor(callback, value) {
|
||||
pronto.check_callback(callback, 'time_limit')
|
||||
var finished = false
|
||||
var requestor_cancel = null
|
||||
var timer_cancel = null
|
||||
|
||||
function cancel(reason) {
|
||||
if (finished) return
|
||||
finished = true
|
||||
if (timer_cancel) {
|
||||
timer_cancel()
|
||||
timer_cancel = null
|
||||
}
|
||||
if (requestor_cancel) {
|
||||
requestor_cancel(reason)
|
||||
requestor_cancel = null
|
||||
}
|
||||
}
|
||||
|
||||
function safe_cancel_requestor(reason) {
|
||||
if (requestor_cancel) {
|
||||
requestor_cancel(reason)
|
||||
requestor_cancel = null
|
||||
}
|
||||
}
|
||||
|
||||
timer_cancel = $_.delay(function() {
|
||||
if (finished) return
|
||||
def reason = make_reason(factory, 'Timeout.', seconds)
|
||||
safe_cancel_requestor(reason)
|
||||
finished = true
|
||||
callback(null, reason)
|
||||
}, seconds)
|
||||
|
||||
function do_request() {
|
||||
requestor_cancel = requestor(function(val, reason) {
|
||||
if (finished) return
|
||||
finished = true
|
||||
if (timer_cancel) {
|
||||
timer_cancel()
|
||||
timer_cancel = null
|
||||
}
|
||||
callback(val, reason)
|
||||
}, value)
|
||||
} disruption {
|
||||
cancel('requestor failed')
|
||||
callback(null, 'requestor failed')
|
||||
}
|
||||
do_request()
|
||||
|
||||
return function(reason) {
|
||||
safe_cancel_requestor(reason)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
var config = {
|
||||
ar_timer: 60,
|
||||
actor_memory:0,
|
||||
@@ -1210,31 +1361,8 @@ function guid(bits)
|
||||
return text(guid,'h')
|
||||
}
|
||||
|
||||
var HEADER = {}
|
||||
|
||||
// takes a function input value that will eventually be called with the current time in number form.
|
||||
$_.clock = function(fn) {
|
||||
actor_mod.clock(_ => {
|
||||
fn(time.number())
|
||||
send_messages()
|
||||
})
|
||||
}
|
||||
|
||||
var underlings = {} // this is more like "all actors that are notified when we die"
|
||||
var overling = null
|
||||
var root = null
|
||||
|
||||
var receive_fn = null
|
||||
var greeters = {} // Router functions for when messages are received for a specific actor
|
||||
|
||||
var enet = use_core('internal/enet')
|
||||
|
||||
var peers = {}
|
||||
var id_address = {}
|
||||
var peer_queue = {}
|
||||
var portal = null
|
||||
var portal_fn = null
|
||||
var enet = use_core('enet')
|
||||
enet = use_core('internal/enet')
|
||||
enet = use_core('enet')
|
||||
|
||||
function peer_connection(peer) {
|
||||
return {
|
||||
@@ -1259,37 +1387,6 @@ function peer_connection(peer) {
|
||||
}
|
||||
}
|
||||
|
||||
// takes a callback function, an actor object, and a configuration record for getting information about the status of a connection to the actor. The configuration record is used to request the sort of information that needs to be communicated. This can include latency, bandwidth, activity, congestion, cost, partitions. The callback is given a record containing the requested information.
|
||||
$_.connection = function(callback, actor, config) {
|
||||
var peer = peers[actor[ACTORDATA].id]
|
||||
if (peer) {
|
||||
callback(peer_connection(peer))
|
||||
return
|
||||
}
|
||||
if (actor_mod.mailbox_exist(actor[ACTORDATA].id)) {
|
||||
callback({type:"local"})
|
||||
return
|
||||
}
|
||||
|
||||
callback()
|
||||
}
|
||||
|
||||
// takes a function input value that will eventually be called with the current time in number form.
|
||||
$_.portal = function(fn, port) {
|
||||
if (portal) {
|
||||
log.error(`Already started a portal listening on ${portal.port()}`)
|
||||
disrupt
|
||||
}
|
||||
if (!port) {
|
||||
log.error("Requires a valid port.")
|
||||
disrupt
|
||||
}
|
||||
log.system(`starting a portal on port ${port}`)
|
||||
portal = enet.create_host({address: "any", port})
|
||||
portal_fn = fn
|
||||
enet_check()
|
||||
}
|
||||
|
||||
function handle_host(e) {
|
||||
var queue = null
|
||||
var data = null
|
||||
@@ -1337,78 +1434,6 @@ function populate_actor_addresses(obj, e) {
|
||||
})
|
||||
}
|
||||
|
||||
// takes a callback function, an actor object, and a configuration record for getting information about the status of a connection to the actor. The configuration record is used to request the sort of information that needs to be communicated. This can include latency, bandwidth, activity, congestion, cost, partitions. The callback is given a record containing the requested information.
|
||||
$_.contact = function(callback, record) {
|
||||
send(create_actor(record), record, callback)
|
||||
}
|
||||
|
||||
// registers a function that will receive all messages...
|
||||
$_.receiver = function receiver(fn) {
|
||||
receive_fn = fn
|
||||
}
|
||||
|
||||
// Holds all messages queued during the current turn.
|
||||
var message_queue = []
|
||||
|
||||
$_.start = function start(cb, program) {
|
||||
if (!program) return
|
||||
|
||||
var id = guid()
|
||||
var oid = $_.self[ACTORDATA].id
|
||||
var startup = {
|
||||
id,
|
||||
overling_id: oid,
|
||||
root_id: root ? root[ACTORDATA].id : null,
|
||||
program,
|
||||
native_mode: native_mode,
|
||||
no_warn: _no_warn,
|
||||
}
|
||||
greeters[id] = cb
|
||||
push(message_queue, { startup })
|
||||
}
|
||||
|
||||
// stops an underling or self.
|
||||
$_.stop = function stop(actor) {
|
||||
if (!actor) {
|
||||
need_stop = true
|
||||
return
|
||||
}
|
||||
if (!is_actor(actor)) {
|
||||
log.error('Can only call stop on an actor.')
|
||||
disrupt
|
||||
}
|
||||
if (is_null(underlings[actor[ACTORDATA].id])) {
|
||||
log.error('Can only call stop on an underling or self.')
|
||||
disrupt
|
||||
}
|
||||
|
||||
sys_msg(actor, {kind:"stop"})
|
||||
}
|
||||
|
||||
// schedules the removal of an actor after a specified amount of time.
|
||||
$_.unneeded = function unneeded(fn, seconds) {
|
||||
actor_mod.unneeded(fn, seconds)
|
||||
}
|
||||
|
||||
// schedules the invocation of a function after a specified amount of time.
|
||||
$_.delay = function delay(fn, seconds) {
|
||||
var _seconds = seconds == null ? 0 : seconds
|
||||
function delay_turn() {
|
||||
fn()
|
||||
send_messages()
|
||||
}
|
||||
var id = actor_mod.delay(delay_turn, _seconds)
|
||||
return function() { actor_mod.removetimer(id) }
|
||||
}
|
||||
|
||||
// causes this actor to stop when another actor stops.
|
||||
var couplings = {}
|
||||
$_.couple = function couple(actor) {
|
||||
if (actor == $_.self) return // can't couple to self
|
||||
couplings[actor[ACTORDATA].id] = true
|
||||
sys_msg(actor, {kind:'couple', from_id: _cell.id})
|
||||
log.system(`coupled to ${actor[ACTORDATA].id}`)
|
||||
}
|
||||
|
||||
function actor_prep(actor, send) {
|
||||
push(message_queue, {actor,send});
|
||||
@@ -1497,8 +1522,6 @@ function send_messages() {
|
||||
message_queue = []
|
||||
}
|
||||
|
||||
var replies = {}
|
||||
|
||||
function send(actor, message, reply) {
|
||||
var send_msg = null
|
||||
var target = null
|
||||
@@ -1547,11 +1570,6 @@ function send(actor, message, reply) {
|
||||
|
||||
stone(send)
|
||||
|
||||
if (!_cell.args.id) _cell.id = guid()
|
||||
else _cell.id = _cell.args.id
|
||||
|
||||
$_.self = create_actor({id: _cell.id})
|
||||
|
||||
// Actor's timeslice for processing a single message
|
||||
function turn(msg)
|
||||
{
|
||||
@@ -1569,12 +1587,6 @@ if (config.actor_memory)
|
||||
if (config.stack_max)
|
||||
js.max_stacksize(config.system.stack_max);
|
||||
|
||||
overling = _cell.args.overling_id ? create_actor({id: _cell.args.overling_id}) : null
|
||||
$_.overling = overling
|
||||
|
||||
root = _cell.args.root_id ? create_actor({id: _cell.args.root_id}) : null
|
||||
if (root == null) root = $_.self
|
||||
|
||||
if (overling) {
|
||||
$_.couple(overling)
|
||||
report_to_overling({type:'greet', actor_id: _cell.id})
|
||||
|
||||
Reference in New Issue
Block a user