Files
cell/scripts/engine.cm

825 lines
22 KiB
Plaintext

(function engine() {
globalThis.cell = prosperon
cell.DOC = Symbol()
var ACTORDATA = cell.hidden.actorsym
var SYSYM = '__SYSTEM__'
var ENETSERVICE = 0.1
var REPLYTIMEOUT = 60 // seconds before replies are ignored
var MOD_EXT = '.cm'
var ACTOR_EXT = '.ce'
globalThis.pi = 3.1415926535897932
function caller_data(depth = 0)
{
var file = "nofile"
var line = 0
var caller = new Error().stack.split("\n")[1+depth]
if (caller) {
var md = caller.match(/\((.*)\:/)
var m = md ? md[1] : "SCRIPT"
if (m) file = m
md = caller.match(/\:(\d*)\)/)
m = md ? md[1] : 0
if (m) line = m
}
return {file,line}
}
cell.args = cell.hidden.init
cell.args ??= {}
cell.id ??= "newguy"
function console_rec(line, file, msg) {
return `[${cell.id.slice(0,5)}] [${file}:${line}]: ${msg}\n`
// time: [${time.text("mb d yyyy h:nn:ss")}]
}
var console_mod = cell.hidden.console
var logs = {}
logs.console = function(msg)
{
var caller = caller_data(4)
console_mod.print(console_rec(caller.line, caller.file, msg))
}
logs.error = function(msg = new Error())
{
var caller = caller_data(4)
if (msg instanceof Error)
msg = msg + "\n" + msg.stack
console_mod.print(console_rec(caller.line,caller.file,msg))
}
logs.system = function(msg) {
msg = "[SYSTEM] " + msg
log.console(msg)
}
function noop() {}
globalThis.log = new Proxy(logs, {
get(target,prop,receiver) {
if (target[prop])
return (...args) => args.forEach(arg => target[prop](arg))
return noop
}
})
// Get hidden modules from cell.hidden before stripping it
var hidden = cell.hidden
var actor_mod = hidden.actor
var wota = hidden.wota
var use_embed = hidden.use_embed
var use_dyn = hidden.use_dyn
var enet = hidden.enet
var nota = hidden.nota
// Strip hidden from cell so nothing else can access it
delete cell.hidden
function disrupt(err)
{
if (overling) {
if (err) {
// with an err, this is a forceful disrupt
var reason = (err instanceof Error) ? err.stack : err
report_to_overling({type:'disrupt', reason})
} else
report_to_overling({type:'stop'})
}
for (var id of underlings)
$_.stop(create_actor({id}))
if (err) {
log.console(err);
if (err.stack)
log.console(err.stack)
}
actor_mod.disrupt()
}
actor_mod.on_exception(disrupt)
var js = use_embed('js')
var io = use_embed('io')
if (!io.exists('.cell')) {
console_mod.print("No cell directory found. Make one.\n");
os.exit(1);
}
var use_cache = {}
var BASEPATH = 'scripts/base' + MOD_EXT
var script = io.slurp(BASEPATH)
var fnname = "base"
script = `(function ${fnname}() { ${script}; })`
js.eval(BASEPATH, script)()
var inProgress = {}
var loadingStack = []
globalThis.use = function use(file, ...args) {
// Check cache first
if (use_cache[file]) {
return use_cache[file]
}
// We'll check for circular dependencies after we determine the path
var path = null
// First check if we're loading from a script and look in its directory
if (loadingStack.length > 0) {
var currentScript = loadingStack[loadingStack.length - 1]
if (currentScript.includes('/')) {
var currentDir = currentScript.substring(0, currentScript.lastIndexOf('/'))
// Try the file name as-is in the current directory
var localPath = currentDir + '/' + file + MOD_EXT
if (io.exists(localPath) && !io.is_directory(localPath)) {
path = localPath
}
}
}
// If not found locally, check the normal path
if (!path && io.exists(file + MOD_EXT) && !io.is_directory(file + MOD_EXT)) {
path = file + MOD_EXT
}
// Check if there's an embedded module
var embed_mod = use_embed(file)
// If no script and no embedded module, error
if (!path && !embed_mod) {
throw new Error(`Module ${file} could not be found`)
}
// If only embedded module exists, return it
if (!path && embed_mod) {
use_cache[file] = embed_mod
return embed_mod
}
// Check for circular dependencies using the resolved path
if (path && loadingStack.includes(path)) {
let cycleIndex = loadingStack.indexOf(path)
let cyclePath = loadingStack.slice(cycleIndex).concat(path)
throw new Error(
`Circular dependency detected while loading "${file}".\n` +
`Module chain: ${loadingStack.join(" -> ")}\n` +
`Cycle specifically: ${cyclePath.join(" -> ")}`
)
}
inProgress[path] = true
loadingStack.push(path)
// Determine the compiled file path in .cell directory
var compiledPath = ".cell/build/" + io.realdir(path) + "/" + path + '.o'
io.mkdir(compiledPath.dir())
// Check if compiled version exists and is newer than source
var useCompiled = false
if (io.exists(compiledPath)) {
var srcStat = io.stat(path)
var compiledStat = io.stat(compiledPath)
if (compiledStat.modtime >= srcStat.modtime) {
useCompiled = true
}
}
var fn
var mod_name = path.name()
if (useCompiled) {
var compiledBlob = io.slurpbytes(compiledPath)
fn = js.compile_unblob(compiledBlob)
fn = js.eval_compile(fn)
} else {
// Compile from source
var script = io.slurp(path)
var mod_script = `(function setup_${mod_name}_module(arg){${script};})`
fn = js.compile(path, mod_script)
// Save compiled version to .cell directory
var compiled = js.compile_blob(fn)
io.slurpwrite(compiledPath, compiled)
fn = js.eval_compile(fn)
}
// Create context - if embedded module exists, script extends it
var context = {}
if (embed_mod)
context.__proto__ = embed_mod
// Call the script - pass embedded module as 'this' if it exists
var ret = fn.call(context, args)
// If script doesn't return anything, check if we have embedded module
if (!ret && embed_mod) {
ret = embed_mod
} else if (!ret) {
throw new Error(`Use must be used with a module, but ${path} doesn't return a value`)
}
loadingStack.pop()
delete inProgress[path]
// Cache the result
use_cache[file] = ret
return ret
}
globalThis.json = use('json')
var time = use('time')
var st_now = time.number()
var shop = use('shop')
var config = shop.load_config()
var default_config = {
ar_timer: 60,
actor_memory:0,
net_service:0.1,
reply_timeout:60
}
config.system ??= {}
config.system.__proto__ = default_config
ENETSERVICE = config.system.net_service
REPLYTIMEOUT = config.system.reply_timeout
globalThis.text = use('text')
// Load actor-specific configuration
function load_actor_config(program) {
// Extract actor name from program path
// e.g., "prosperon/_sdl_video" or "extramath/spline"
var actor_name = program
if (program.includes('/')) {
actor_name = program
}
if (config.actors && config.actors[actor_name]) {
// Merge actor config into cell.args
for (var key in config.actors[actor_name]) {
log.console(`setting ${key}`)
cell.args[key] = config.actors[actor_name][key]
}
}
}
var blob = use('blob')
var blob_stone = blob.prototype.stone
var blob_stonep = blob.prototype.stonep;
delete blob.prototype.stone;
delete blob.prototype.stonep;
function deepFreeze(object) {
if (object instanceof blob)
blob_stone.call(object);
// Retrieve the property names defined on object
var propNames = Object.keys(object);
// Freeze properties before freezing self
for (var name of propNames) {
var value = object[name];
if ((value && typeof value === "object") || typeof value === "function") {
deepFreeze(value);
}
}
return Object.freeze(object);
}
globalThis.stone = deepFreeze
stone.p = function(object)
{
if (object instanceof blob)
return blob_stonep.call(object)
return Object.isFrozen(object)
}
/*
When handling a message, the message appears like this:
{
type: type of message
- contact: used for contact messages
- stop: used to issue stop command
- etc
reply: ID this message will respond to (callback saved on the actor)
replycc: the actor that is waiting for the reply
target: ID of the actor that's supposed to receive the message. Only added to non direct sends (out of portals)
return: reply ID so the replycc actor can know what callback to send the message to
data: the actual content of the message
}
actors look like
{
id: the GUID of this actor
address: the IP this actor can be found at
port: the port of the IP the actor can be found at
}
*/
var util = use('util')
var crypto = use('crypto')
var HEADER = Symbol()
function create_actor(desc = {id:util.guid()}) {
var actor = {}
actor[ACTORDATA] = desc
return actor
}
var $_ = create_actor()
$_.random = crypto.random
$_.random[cell.DOC] = "returns a number between 0 and 1. There is a 50% chance that the result is less than 0.5."
$_.random_fit = crypto.random_fit
$_.clock = function(fn) {
$_.delay(_ => {
fn(time.number())
}, 0)
}
$_.clock[cell.DOC] = "takes a function input value that will eventually be called with the current time in number form."
var underlings = new Set() // this is more like "all actors that are notified when we die"
var overling = undefined
var root = undefined
// Don't make $_ global - it should only be available to actor scripts
var receive_fn = undefined
var greeters = {} // Router functions for when messages are received for a specific actor
globalThis.is_actor = function is_actor(actor) {
return actor[ACTORDATA]
}
function peer_connection(peer) {
return {
latency: peer.rtt,
bandwidth: {
incoming: peer.incoming_bandwidth,
outgoing: peer.outgoing_bandwidth
},
activity: {
last_sent: peer.last_send_time,
last_received: peer.last_receive_time
},
mtu: peer.mtu,
data: {
incoming_total: peer.incoming_data_total,
outgoing_total: peer.outgoing_data_total,
reliable_in_transit: peer.reliable_data_in_transit
},
latency_variance: peer.rtt_variance,
packet_loss: peer.packet_loss,
state: peer.state
}
}
$_.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()
}
$_.connection[cell.DOC] = "The connection function 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."
var peers = {}
var id_address = {}
var peer_queue = new WeakMap()
var portal = undefined
var portal_fn = undefined
$_.portal = function(fn, port) {
if (portal) throw new Error(`Already started a portal listening on ${portal.port}`)
if (!port) throw new Error("Requires a valid port.")
log.system(`starting a portal on port ${port}`)
portal = enet.create_host({address: "any", port})
portal_fn = fn
}
$_.portal[cell.DOC] = "A portal is a special actor with a public address that performs introduction services. It listens on a specified port for contacts by external actors that need to acquire an actor object. The function will receive the record containing the request. The record can have a reply sent through it. A portal can respond by beginning a new actor, or finding an existing actor, or by forwarding the contact message to another actor. This is how distributed Misty networks are bootstrapped. The portal function returns null."
function handle_host(e) {
switch (e.type) {
case "connect":
log.system(`connected a new peer: ${e.peer.address}:${e.peer.port}`)
peers[`${e.peer.address}:${e.peer.port}`] = e.peer
var queue = peer_queue.get(e.peer)
if (queue) {
for (var msg of queue) e.peer.send(nota.encode(msg))
log.system(`sent ${json.encode(msg)} out of queue`)
peer_queue.delete(e.peer)
}
break
case "disconnect":
peer_queue.delete(e.peer)
for (var id in peers) if (peers[id] === e.peer) delete peers[id]
log.system('portal got disconnect from ' + e.peer.address + ":" + e.peer.port)
break
case "receive":
var data = nota.decode(e.data)
if (data.replycc && !data.replycc.address) {
data.replycc[ACTORDATA].address = e.peer.address
data.replycc[ACTORDATA].port = e.peer.port
}
function populate_actor_addresses(obj) {
if (typeof obj !== 'object' || obj === null) return
if (obj[ACTORDATA] && !obj[ACTORDATA].address) {
obj[ACTORDATA].address = e.peer.address
obj[ACTORDATA].port = e.peer.port
}
for (var key in obj) {
if (obj.hasOwnProperty(key)) {
populate_actor_addresses(obj[key])
}
}
}
if (data.data) populate_actor_addresses(data.data)
turn(data)
break
}
}
$_.contact = function(callback, record) {
send(create_actor(record), record, callback)
}
$_.contact[cell.DOC] = `The contact function sends a message to a portal on another machine to obtain an actor object.
The callback is a function with a actor input and a reason input. If successful, actor is bound to an actor object. If not successful, actor is null and reason may contain an explanation.`
$_.receiver = function receiver(fn) {
receive_fn = fn
}
$_.receiver[cell.DOC] = "registers a function that will receive all messages..."
$_.start = function start(cb, program, ...args) {
if (!program) return
var id = util.guid()
if (args.length === 1 && Array.isArray(args[0]))
args = args[0]
var startup = {
id,
overling: $_,
root,
arg: args,
program
}
greeters[id] = cb
actor_mod.createactor(startup)
}
$_.stop = function stop(actor) {
if (!actor) {
need_stop = true
return
}
if (!is_actor(actor))
throw new Error('Can only call stop on an actor.')
if (!underlings.has(actor[ACTORDATA].id))
throw new Error('Can only call stop on an underling or self.')
sys_msg(actor, "stop")
}
$_.stop[cell.DOC] = "The stop function stops an underling."
$_.unneeded = function unneeded(fn, seconds) {
actor_mod.unneeded(fn, seconds)
}
$_.delay = function delay(fn, seconds) {
function delay_turn() {
fn()
send_messages()
}
var id = actor_mod.delay(delay_turn, seconds)
return function() { actor_mod.removetimer(id) }
}
$_.delay[cell.DOC] = "used to schedule the invocation of a function..."
var couplings = new Set()
$_.couple = function couple(actor) {
if (actor === $_) return // can't couple to self
couplings.add(actor[ACTORDATA].id)
sys_msg(actor, {kind:'couple', from: $_})
log.system(`coupled to ${actor}`)
}
$_.couple[cell.DOC] = "causes this actor to stop when another actor stops."
function actor_prep(actor, send) {
message_queue.push({actor,send});
}
// Send a message immediately without queuing
function actor_send_immediate(actor, send) {
try {
actor_send(actor, send);
} catch (err) {
log.error("Failed to send immediate message:", err);
}
}
function actor_send(actor, message) {
if (actor[HEADER] && !actor[HEADER].replycc) // attempting to respond to a message but sender is not expecting; silently drop
return
if (!is_actor(actor) && !is_actor(actor.replycc)) throw new Error(`Must send to an actor object. Attempted send to ${json.encode(actor)}`)
if (typeof message !== 'object') throw new Error('Must send an object record.')
// message to self
if (actor[ACTORDATA].id === cell.id) {
if (receive_fn) receive_fn(message.data)
return
}
// message to actor in same flock
if (actor[ACTORDATA].id && actor_mod.mailbox_exist(actor[ACTORDATA].id)) {
actor_mod.mailbox_push(actor[ACTORDATA].id, message)
return
}
if (actor[ACTORDATA].address) {
if (actor[ACTORDATA].id)
message.target = actor[ACTORDATA].id
else
message.type = "contact"
var peer = peers[actor[ACTORDATA].address + ":" + actor[ACTORDATA].port]
if (!peer) {
if (!portal) {
log.system(`creating a contactor ...`)
portal = enet.create_host({address:"any"})
log.system(`allowing contact to port ${portal.port}`)
}
log.system(`no peer! connecting to ${actor[ACTORDATA].address}:${actor[ACTORDATA].port}`)
peer = portal.connect(actor[ACTORDATA].address, actor[ACTORDATA].port)
peer_queue.set(peer, [message])
} else {
peer.send(nota.encode(message))
}
return
}
log.system(`Unable to send message to actor ${json.encode(actor[ACTORDATA])}`)
}
// Holds all messages queued during the current turn.
var message_queue = []
var need_stop = false
function send_messages() {
for (var msg of message_queue)
actor_send(msg.actor,msg.send)
message_queue.length = 0
if (need_stop)
disrupt()
}
var replies = {}
globalThis.send = function send(actor, message, reply) {
if (typeof actor !== 'object')
throw new Error('Must send to an actor object. Provided: ' + actor);
if (typeof message !== 'object')
throw new Error('Message must be an object')
var send = {type:"user", data: message}
if (actor[HEADER] && actor[HEADER].replycc) {
var header = actor[HEADER]
if (!header.replycc || !is_actor(header.replycc))
throw new Error(`Supplied actor had a return, but it's not a valid actor! ${json.encode(actor[HEADER])}`)
actor = header.replycc
send.return = header.reply
}
if (reply) {
var id = util.guid()
replies[id] = reply
$_.delay(_ => {
if (replies[id]) {
replies[id](undefined, "timeout")
delete replies[id]
}
}, REPLYTIMEOUT)
send.reply = id
send.replycc = $_
}
// Instead of sending immediately, queue it
actor_prep(actor,send);
}
stone(send)
if (!cell.args.id) cell.id = util.guid()
else cell.id = cell.args.id
$_[ACTORDATA].id = cell.id
// Actor's timeslice for processing a single message
function turn(msg)
{
handle_message(msg)
send_messages()
}
load_actor_config(cell.args.program)
actor_mod.register_actor(cell.id, turn, cell.args.main, config.system.ar_timer)
if (config.system.actor_memory)
js.mem_limit(config.system.actor_memory)
if (config.system.stack_max)
js.max_stacksize(config.system.stack_max);
overling = cell.args.overling
root = cell.args.root
root ??= $_
if (overling) {
$_.couple(overling) // auto couple to overling
report_to_overling({type:'greet', actor: $_})
}
// sys messages are always dispatched immediately
function sys_msg(actor, msg)
{
actor_send(actor, {[SYSYM]:msg})
}
// messages sent to here get put into the cb provided to start
function report_to_overling(msg)
{
if (!overling) return
sys_msg(overling, {kind:'underling', message:msg, from: $_})
}
if (!cell.args.program)
os.exit(1)
function handle_actor_disconnect(id) {
var greeter = greeters[id]
if (greeter) {
greeter({type: "stopped", id})
delete greeters[id]
}
log.system(`actor ${id} disconnected`)
if (couplings.has(id)) disrupt("coupled actor died") // couplings now disrupts instead of stop
}
function handle_sysym(msg)
{
switch(msg.kind) {
case 'stop':
disrupt("got stop message")
break
case 'underling':
var from = msg.from
var greeter = greeters[from[ACTORDATA].id]
if (greeter) greeter(msg.message)
if (msg.message.type == 'disrupt')
underlings.delete(from[ACTORDATA].id)
break
case 'contact':
if (portal_fn) {
var letter2 = msg.data
letter2[HEADER] = msg
delete msg.data
portal_fn(letter2)
} else throw new Error('Got a contact message, but no portal is established.')
break
case 'couple': // from must be notified when we die
var from = msg.from
underlings.add(from[ACTORDATA].id)
log.system(`actor ${from} is coupled to me`)
break
}
}
function handle_message(msg) {
if (msg[SYSYM]) {
handle_sysym(msg[SYSYM], msg.from)
return
}
switch (msg.type) {
case "user":
var letter = msg.data // what the sender really sent
Object.defineProperty(letter, HEADER, {
value: msg, enumerable: false
})
Object.defineProperty(letter, ACTORDATA, { // this is so is_actor === true
value: { reply: msg.reply }, enumerable: false
})
if (msg.return) {
var fn = replies[msg.return]
if (fn) fn(letter)
delete replies[msg.return]
return
}
if (receive_fn) receive_fn(letter)
return
case "stopped":
handle_actor_disconnect(msg.id)
break
}
};
function enet_check()
{
if (portal) portal.service(handle_host)
$_.delay(enet_check, ENETSERVICE);
}
// enet_check();
// Finally, run the program
actor_mod.setname(cell.args.program)
var prog = null
var progPath = cell.args.program
if (io.exists(progPath + ACTOR_EXT) && !io.is_directory(progPath + ACTOR_EXT)) {
prog = progPath + ACTOR_EXT
} else if (io.exists(progPath) && io.is_directory(progPath)) {
// First check for folder's name as a file
var folderName = progPath.split('/').pop()
var folderNamePath = progPath + '/' + folderName + ACTOR_EXT
if (io.exists(folderNamePath) && !io.is_directory(folderNamePath)) {
prog = folderNamePath
} else {
// Fall back to main.ce
var mainPath = progPath + '/main' + ACTOR_EXT
if (io.exists(mainPath) && !io.is_directory(mainPath)) {
prog = mainPath
}
}
}
if (!prog)
throw new Error(cell.args.program + " not found.");
var progDir = io.realdir(prog) + "/" + prog.substring(0, prog.lastIndexOf('/'))
io.mount(progDir.replace(/\/+$/, ''))
var progContent = io.slurp(prog)
var prog_script = `(function ${cell.args.program.name()}_start($_, arg) { var args = arg; ${progContent} })`
// queue up its first turn instead of run immediately
var startfn = js.eval(cell.args.program, prog_script);
$_.delay(_ => {
var val = startfn($_, cell.args.arg);
if (val)
throw new Error('Program must not return anything');
}, 0)
})()