mod COPERATIONAL is

protecting BOOL .
protecting STRING .
protecting NATUTIL .

including CCOMMON .

var dseq seq' seq'' : Nat .
var index newindex oldindex oldindex' transindex : Nat .
var nat nat' : Nat .
var natlist natlist' : NatList .
var natset natset' : NatSet .
var localmsgs : NatSet .
var dseqs seqs : NatSet .
var proc proc' : Proc .
var member : Proc .
var procset procset' : ProcSet .
var destset destset' : ProcSet .
var everybody : ProcSet .
var members members' newmembers transmembers transmembers' : ProcSet .
var oldmembers oldmembers' : ProcSet .
var othernewmembers : ProcSet .
var comp : ProcSet .
var comps : ProcSetSet .
var conf conf' oldconf newconf transconf regconf : Conf .
var mode : Mode .
var data : Data .
var message message'  : Message .
var src dest : Proc .
var knownmsgs : NatSet .
var ackedseq seq : Nat .
var transset : ProcSet .
var ackflag : Bool .
var messagelist messagelist' : MessageList .
var delivered alldelivered : MessageList .
var messageset messageset' : MessageSet .
var received : MessageSet .
var acked : MessageSet .
var constraint constraint' : Constraint .
var constraintset constraintset' : ConstraintSet .
var constraints : ConstraintSet .
var event event' : Event .
var eventlist eventlist' : EventList .
var events events' : EventList .
var reachable : ProcSet .
var broadcast : Broadcast .
var broadcastset : BroadcastSet .
var trans : Bool .


--- broadcasting equations
ceq  sBroadcastSet(broadcast((destset destset'),message)) =
     sBroadcastSet(broadcast(destset,message)) 
     sBroadcastSet(broadcast(destset',message))
     if destset =/= eProcSet and destset' =/= eProcSet .

eq   sBroadcastSet(broadcast(eProcSet,message)) =
     eBroadcastSet .


--- receiving and losing messages
crl ENABLED(RECEIVE(dest,seq))
    sent(src,(sBroadcastSet(broadcast) broadcastset))
    received(dest,messageset') 
    reachable(src,reachable) 
    =>
    PERFORMED(RECEIVE(dest,seq))
    received(dest,(messageset' sMessageSet(message)))
    sent(src,broadcastset)
    reachable(src,reachable)
    if broadcast(sProcSet(dest),message) := broadcast /\
       contains(reachable,dest) /\ seq(message) == seq .

crl ENABLED(LOSE(dest,seq))
    sent(src,(sBroadcastSet(broadcast) broadcastset))
    received(dest,messageset') 
    reachable(src,reachable) 
    =>
    PERFORMED(LOSE(dest,seq))
    received(dest,messageset')
    sent(src,broadcastset)
    reachable(src,reachable)
    if broadcast(sProcSet(dest),message) := broadcast /\
       not(contains(reachable,dest)) /\ seq(message) == seq .


--- An earlier version had a  condition (src =/= dest) making sure that 
--- a process can never lose messages that it sent to itself. But
--- it seems that this can happen when a process crashes (in this
--- case is reachable is empty).

--- Note that this is not sufficient to guarantee receipt. To
--- this end, we additionally need to make sure that a configuration
--- change cannot happen before such a message is actually received.

--- processing multicast request

--- As in Spread we broadcast messages to everybody, but only the
--- processes in the current configuration can deliver it.

rl  operational(proc)
    network(everybody)
    localconf(proc,conf)
    localmsgs(proc,conf,localmsgs)
    knownmsgs(proc,conf,knownmsgs)
    freshseq(seq)
    delivered(proc,messagelist)
    alldelivered(proc,messagelist')
    causalorder(conf,constraints)
    sent(proc,broadcastset)
    ENABLED(SEND(proc,seq))
    multicast-req(proc,mode,data) 
    =>
    operational(proc)
    network(everybody)
    localconf(proc,conf)
    localmsgs(proc,conf, localmsgs sNatSet(seq))
    knownmsgs(proc,conf, knownmsgs sNatSet(seq))
    freshseq((s seq))
    delivered(proc,messagelist)
    alldelivered(proc,messagelist')
    PERFORMED(SEND(proc,seq))
    causalorder(conf,(addConstraints(constraints,proc,localmsgs,knownmsgs,seq,mode)))
    sent(proc, broadcastset sBroadcastSet(broadcast(everybody,datamsg(proc,mode,conf,seq,knownmsgs,data))))
    multicast-ack(proc) .


--- delivery of non-safe data messages

crl operational(proc)
    localconf(proc,conf)
    delivered(proc,delivered)
    localmsgs(proc,conf, localmsgs)
    knownmsgs(proc,conf,knownmsgs)
    received(proc,(sMessageSet(message) received))
    alldelivered(proc,alldelivered)
    causalorder(conf,constraints) 
    totalorder(conf,events) 
    =>
    operational(proc)
    localconf(proc,conf)
    localmsgs(proc,conf, localmsgs)
    knownmsgs(proc,conf, knownmsgs knownmsgs(message) sNatSet(seq(message)))
    received(proc,received)
    delivered(proc,(delivered sMessageList(message)))
    alldelivered(proc,(alldelivered sMessageList(message)))
    causalorder(conf,constraints)
    totalorder(conf,addEvent(events,src(message),seq(message),mode(message)))
    if deliverable(proc,conf,received,alldelivered,message,constraints,events) and 
       not(safe(mode(message))) .


--- sending acknowlegements for safe messages (without delivering)

crl operational(proc)
    localconf(proc,conf)
    delivered(proc,delivered)
    localmsgs(proc,conf, localmsgs)
    freshseq(seq)
    received(proc,(sMessageSet(message) received))
    alldelivered(proc,alldelivered)
    acked(proc, acked)
    causalorder(conf,constraints) 
    totalorder(conf,events) 
    sent(proc, broadcastset) 
    =>
    operational(proc)
    localconf(proc,conf)
    freshseq((s seq))
    received(proc,(sMessageSet(message) received))
    acked(proc,(sMessageSet(message) acked))
    delivered(proc,delivered)
    localmsgs(proc,conf, localmsgs)
    alldelivered(proc,alldelivered)
    causalorder(conf,constraints)
    totalorder(conf,events)
    sent(proc, broadcastset sBroadcastSet(broadcast(members(conf),ackmsg(proc,conf,seq,seq(message)))))
    if deliverable(proc,conf,received,alldelivered,message,constraints,events) and
       safe(mode(message)) and not(contains(acked,message)) .


--- delivering safe messages for which all required acks have been received

crl operational(proc)
    localconf(proc,conf)
    delivered(proc,delivered)
    localmsgs(proc,conf, localmsgs)
    knownmsgs(proc,conf,knownmsgs)
    received(proc,sMessageSet(message) received)
    alldelivered(proc,alldelivered)
    acked(proc, acked)
    causalorder(conf,constraints) 
    totalorder(conf,events) 
    =>
    operational(proc)
    localconf(proc,conf)
    localmsgs(proc,conf, localmsgs)
    knownmsgs(proc,conf, knownmsgs knownmsgs(message) sNatSet(seq(message)))
    received(proc,received)
    acked(proc, acked)
    delivered(proc,(delivered sMessageList(message)))
    alldelivered(proc,(alldelivered sMessageList(message)))
    causalorder(conf,constraints)
    totalorder(conf,addEvent(events,src(message),seq(message),mode(message)))
    if deliverablesafe(proc,conf,received,alldelivered,message,constraints,events,false) and
       safe(mode(message)) and contains(acked,message) .


--- remove messages sent out in previous regular configurations 
--- (this is just for garbage collection, and not really needed)

crl operational(proc)
    localconf(proc,conf)
    received(proc,(sMessageSet(message) received)) 
    =>
    operational(proc)
    localconf(proc,conf)
    received(proc,received)
    if conf(message) =/= reg(conf) .

crl operational(proc)
    localconf(proc,conf)
    acked(proc,(sMessageSet(message) acked)) 
    =>
    operational(proc)
    localconf(proc,conf)
    acked(proc,acked)
    if conf(message) =/= reg(conf) .

endm