#lang ivy1.5

## This file contains a first attempt at refining the network model to
## a sort of toy TCP. 

## Two conjectures were added about the messages in
## the lower (UDP) network layer (that is, they must obey the same
## invariants are the message in the transport layer messages queues.

## A bug was observed in the lower network layer specification, which
## did not promise to deliver a message only to its destination.

## Forgot to include seq_nums object in iso_n

include std

################################################################################
#
# This is the abstract specification of the sharded hash table protocol. It defines
# the consistency reference model (reference)
#
################################################################################


################################################################################
#
# Global type declarations
#
################################################################################


########################################
# type of local time

type ltime
instantiate total_order(ltime)

########################################
# type of hash table data

type data

########################################
# type of hash table keys

type key
instantiate total_non_strict_order(key)

object keys = {
    action max returns (x:key) = {
	assume X <= x
    }
    action next(x:key) returns (y:key) = {
	assert exists Z. ~(Z <= x);
	assume x <= y & x ~= y & (X <= y & X ~= y -> X <= x)
    }
    delegate next
}

axiom 0 <= K:key

########################################
# type of message types

type mtype = {request_t, reply_t, delegate_t, ack_t}

########################################
# type of hash table ops

type otype = {read, write}

########################################
# id's of protocol agents

type id

########################################
# Shards contain a subset of a map from a low to a hi key
#

type shard
function shard_lo(S:shard) : key
function shard_hi(S:shard) : key
function shard_value(S:shard,K:key) : data

########################################
# A request message has a source, key, op
# data and ltime
#

type req
function req_src(R:req) : id
function req_key(R:req) : key
function req_otype(R:req) : otype
function req_data(R:req) : data
function req_ltime(R:req) : ltime

# request messages are extensional

axiom
req_src(R) = req_src(S) &
req_key(R) = req_key(S) &
req_otype(R) = req_otype(S) &
req_data(R) = req_data(S) &
req_ltime(R) = req_ltime(S) &
req_src(R) = req_src(S)
-> R = S

object reqs = {
    action make(src:id,k:key,op:otype,d:data,lt:ltime) returns(res:req) = {
	assume req_src(R) = src &
	req_key(R) = k &
	req_otype(R) = op &
	req_data(R) = d &
	req_ltime(R) = lt
    }
}

########################################
# structure of hash table events to serialize
module hashev = {
    individual type_ : otype
    individual key_ : key
    individual data_  : data # data for write and cas
    individual id_ : id      # process id of op
    relation serialized      # this event has happened

    init ~serialized
}

################################################################################
#
# Reference specification
#
# This module describes a set of memory events ordered by local
# time. It provides actions that assign global time to events in
# increasing order. The pre-conditions of these actions enforce the
# consistency model (that is, what orderings in local time must be
# preserved in global time). 
#
################################################################################

object reference = {

    ########################################
    # memory events by local time

    instantiate evs(T:ltime) : hashev 


    ########################################
    # global memory state 
    # obeys partial function axiom

    individual hash(A:key) : data

    # An aribtrary init value
    init hash(A) = 0
    
    ########################################
    # serialize an event lt at current global time. The "id"
    # parameter tells us what process is serializing the event.
    # 

    action serialize(lt:ltime, id_:id) = {

	assert ~evs(lt).serialized;

	# serialize at current global time
	evs(lt).serialized := true;

	# update the global memory state
	local a : key, d : data {
            a := evs(lt).key_;
	    d := evs(lt).data_;
	    if evs(lt).type_ = read {
		evs(lt).data_ := hash(a)
	    }
	    else { # write
		hash(a) := d
	    }			
	}
    }     
    delegate serialize

    ########################################
    # Note: if we want linearizability, we can add
    # begin and end transaction events to this interface

}

################################################################################
#
# Specification of hash tables
#

module hash_top(key,value) = {

    # The server's hash table

    function hash(X:key) : data
    init hash(X) = 0

    action set_(k:key, v:value) = {
	hash(k) := v
    }

    action get(k:key) returns (v:value) = {
	assume v = hash(k)
    }

    # This extracts a shard from a hash table

    action extract_(lo:key,hi:key) returns(res:shard) = {
	assume shard_lo(res) = lo;
	assume shard_hi(res) = hi;
	assume (lo <= X & X <= hi) -> shard_value(res,X) = hash(X)
    }

    # This incorporates a shard into a hash table

    action incorporate(s:shard) = {
	hash(K) := shard_value(s,K) if (shard_lo(s) <= K & K <= shard_hi(s)) else hash(K)
    }

}



################################################################################
#
# Specification of delegation map
#

module delegation_map_spec = {

    # The delegation map has to be specified with a relation, since as
    # a function it would be unstratified.

    relation map(K:key, X:id)
    init map(K,X) <-> X = 0

    action set_(lo:key,hi:key,dst:id) = {
	map(K,X) := (X = dst) & (lo <= K & K <= hi)
	            | map(K,X) & ~(lo <= K & K <= hi)
    }
#    execute set before impl.set

    action get(k:key) returns (val:id) = {
	assert map(k,val)   # assume not assign because this code is ghost
    }
#    execute get after impl.get

    conjecture map(K,V) & map(K,W) -> V = W

}


################################################################################
#
# Implementation of delegation map
#

module delegation_map() = {

    function entries(K:key) :id
    init entries(0) = 0
    instance has : ordered_set(key)

    action set_(lo:key,hi:key,dst:id) = {
	if lo <= hi {
	    if hi ~= keys.max {
		local nkey:key, nid:id {
		    nkey := keys.next(hi);
		    nid := entries(has.get_glb(nkey));
		    entries(nkey) := nid;
		    call has.insert(nkey)
		}
	    };
	    call has.erase(lo,hi);
	    entries(lo) := dst;
	    call has.insert(lo)
	}
    }

    action get(k:key) returns (val:id) = {
	val := entries(has.get_glb(k))
    }

    # The value of every key between K and its successor is the same as K's value.

    conjecture has.s(K) & has.succ.map(K,L) & K <= M & (L = 0 | ~(L <= M)) -> dma.map(me,M,entries(K))

    # We always have an entry for key zero. 

    conjecture has.s(0)
}

################################################################################
# 
# This is the top-level server description
#
# "ref" is the reference object
# "me" is the server's id
# net is the network

module server_top(ref,me,neti) = {

    # The server's hash table
    
    instance hash : hash_top(key,data)
    
    # The server's delegation map

    instance dm : delegation_map
    private dm

    action handle_request(rq:req) = {
	local src:id, k:key, op:otype, d:data, lt:ltime, ow:id {
	    src := req_src(rq);
	    k := req_key(rq);
	    op := req_otype(rq);
	    d := req_data(rq);
	    lt := req_ltime(rq);
	    ow := dm.get(k);
	    if ow = me {
		call ref.serialize(lt,me);  # this is ghost!
		if op = read {
		    d := hash.get(k)
		}
		else {
		    call hash.set_(k,d)
		};
		call neti.send_reply(me, src, d, lt)
	    } else {
		call neti.send_request(me, ow, rq)  # forward request
	    }
	}
    }

    action generate_request(rq:req) = {
	call handle_request(rq)
    }

    action recv_request(rq:req) = {
	call handle_request(rq)
    }

    action shard(dst:id, lo:key, hi:key) = {
	assume dst ~= me;
	assume lo <= K & K <= hi -> dm.get(K:key) = me;
	call dm.set_(lo,hi,dst);
	call neti.send_delegate(me,dst,hash.extract_(lo,hi))
    }

    action recv_delegate(s:shard) = {
	call dm.set_(shard_lo(s),shard_hi(s),me);
	call hash.incorporate(s)
    }


    # If I own this key, then my hash table data matches the reference
    conjecture dma(me).map(K,me) -> hash.hash(K) = ref.hash(K)

    # If I own this key, then no one else does
    conjecture dma(me).map(K,me) & X ~= me -> ~dma(X).map(K,X)

    # If I own this key, then no delegated shard does
    conjecture dma(me).map(K,me) -> ~(net.delegated(X,S) & shard_lo(S) <= K & K <= shard_hi(S))

    # No two delegated shards have keys in common
    conjecture net.delegated(X,S) & shard_lo(S) <= K & K <= shard_hi(S)
               & net.delegated(X1,S1) & shard_lo(S1) <= K & K <= shard_hi(S1)
               -> X = X1 & S = S1

    # Forwarded requests have correct operations relative to the reference

    conjecture net.requested(D,R) & L = req_ltime(R)->
               (req_key(R) = ref.evs(L).key_ &
                req_otype(R) = ref.evs(L).type_ &
                (req_otype(R) = write -> req_data(R) = ref.evs(L).data_))

    # All forwarded requests have been generated

    conjecture net.requested(D,R) -> io_spec.generated(req_ltime(R)) & ~ref.evs(req_ltime(R)).serialized

    # No two forwarded requests with the same ltime

    conjecture net.requested(D1,R1) & net.requested(D2,R2) & req_ltime(R1) = req_ltime(R2)
               -> D1 = D2 & R1 = R2

    # Delegated shards have correct data

    conjecture net.delegated(X,S) & shard_lo(S) <= K & K <= shard_hi(S) -> shard_value(S,K) = ref.hash(K)

    conjecture ref.evs(T).serialized -> io_spec.generated(T)
}


################################################################################
#
# Specification of network top layer
#
# Currently, delivery is out-of-order non-duplicating

module net_top(ref,proto,neti) = {

    relation requested(D:id,R:req)
    relation delegated(D:id,S:shard)

    init ~requested(D,R)
    init ~delegated(D,S)


    action send_request(src:id,dst:id,rq:req) = {
	assert ~requested(dst,rq);
	requested(dst,rq) := true
    }
    execute send_request before neti.send_request

    action send_delegate(src:id,dst:id,s:shard) = {
	assert ~delegated(dst,s);
	delegated(dst,s) := true
    }
    execute send_delegate before neti.send_delegate

    action recv_request(dst:id,rq:req) = {
	assert requested(dst,rq);
	requested(dst,rq) := false
    }
    execute recv_request before proto.recv_request


    action recv_delegate(dst:id,s:shard) = {
	assume delegated(dst,s);
	delegated(dst,s) := false
    }
    execute recv_delegate before proto.recv_delegate
}

type seq_num
instantiate total_non_strict_order(seq_num)

object seq_nums = {
    action next(seq:seq_num) returns (res:seq_num) = {
	# This was inconsistent!
	assume seq <= res & seq ~= res & (X <= res & X ~= res -> X <= seq)
    }
}

type net_msg

function net_msg_mtype(M:net_msg) : mtype
function net_msg_src(M:net_msg) : id
function net_msg_dst(M:net_msg) : id
function net_msg_req(M:net_msg) : req
function net_msg_seq_num(M:net_msg) : seq_num
function net_msg_shard(M:net_msg) : shard


object encoder = {

    action mk_req(src:id,dst:id,rq:req,seq:seq_num) returns (msg:net_msg) = {
	assume net_msg_src(msg) = src
	  & net_msg_dst(msg) = dst
	  & net_msg_req(msg) = rq
	  & net_msg_seq_num(msg) = seq
	  & net_msg_mtype(msg) = request_t
    }

    action mk_delegate(src:id,dst:id,sh:shard,seq:seq_num) returns (msg:net_msg) = {
	assume net_msg_src(msg) = src
	  & net_msg_dst(msg) = dst
	  & net_msg_shard(msg) = sh
	  & net_msg_seq_num(msg) = seq
	  & net_msg_mtype(msg) = delegate_t
    }

    action mk_ack(src:id,dst:id,seq:seq_num) returns (msg:net_msg) = {
	assume net_msg_src(msg) = src
	  & net_msg_dst(msg) = dst
	  & net_msg_seq_num(msg) = seq
	  & net_msg_mtype(msg) = ack_t
    }
    
}

module message_queue_top(me) = {

    relation contents(M:net_msg)
    init ~contents(M)

    action enqueue(msg:net_msg) = {
	assert ~contents(msg);
	contents(msg) := true
    }

    action empty returns (res:bool) = {
	assume contents(M) -> ~res;
	assume ~res -> exists M. contents(M)
    }

    action pick_one returns (res:net_msg) = {
	assume ~contents(res) -> ~contents(M)
    }
    
    action delete_all(seq:seq_num) = {
	contents(M) := contents(M) & ~(net_msg_seq_num(M) <= seq)
    }

}

module net_impl(ref,proto,me,lower) = {

    instance mq(D:id) : message_queue_top(D)
    individual send_seq(S:id) : seq_num
    individual recv_seq(S:id) : seq_num

    init recv_seq(S) = 0 & send_seq(S) = 0

    action send_request(dst:id,rq:req) = {
	local msg : net_msg, seq : seq_num {
	    msg := encoder.mk_req(me,dst,rq,send_seq(dst));
	    send_seq(dst) := seq_nums.next(send_seq(dst));
	    call mq(dst).enqueue(msg);
	    call lower.send_msg(msg)
        }
    }

    action send_delegate(dst:id,sh:shard) = {
	local msg : net_msg, seq : seq_num {
	    msg := encoder.mk_delegate(me,dst,sh,send_seq(dst));
	    send_seq(dst) := seq_nums.next(send_seq(dst));
	    call mq(dst).enqueue(msg);
	    call lower.send_msg(msg)
        }
    }

    action recv_msg(msg:net_msg) = {
	local src:id,seq:seq_num {
	    seq := net_msg_seq_num(msg);
	    src := net_msg_src(msg);
	    if seq <= recv_seq(src) & net_msg_mtype(msg) ~= ack_t  {
		call lower.send_msg(encoder.mk_ack(me,src,seq))
	    };
	    if seq = recv_seq(src) {
		recv_seq(src) := seq_nums.next(recv_seq(src));
		if net_msg_mtype(msg) = request_t {
		    call proto.recv_request(me,net_msg_req(msg))
		}
		else if net_msg_mtype(msg) = delegate_t {
		    call proto.recv_delegate(me,net_msg_shard(msg))
		}
		else if net_msg_mtype(msg) = ack_t {
		    call mq(src).delete_all(seq)
		}
	    }
	}
    }

    action timeout = {
	if exists D:id,M:net_msg. mq(D).contents(M) {
	    local dst:id, msg:net_msg, b:bool {
		b := mq(dst).empty;
		assume ~b;
		msg := mq(dst).pick_one;
		call lower.send_msg(msg)
	    }
	}
    }

    action send_reply(dst:id, d:data, lt:ltime) = {
	# for now, do nothing, but we verify data correct
    }

    # If I have a request message for D enqueued and if its sequence number is
    # >= D's receive sequence number, then the message is pending.

    conjecture mq(D).contents(M) & neti(D).recv_seq(me) <= net_msg_seq_num(M)
       & net_msg_mtype(M) = request_t -> net.requested(D,net_msg_req(M))

    # If I have a delegate message for D enqueued and if its sequence number is
    # >= D's receive sequence number, then the message is pending.

    conjecture mq(D).contents(M) & neti(D).recv_seq(me) <= net_msg_seq_num(M)
       & net_msg_mtype(M) = delegate_t -> net.delegated(D,net_msg_shard(M))

    # A given request cannot occur twice in the network

    conjecture neti(S1).mq(D).contents(M1) & neti(D).recv_seq(S1) <= net_msg_seq_num(M1)
       & neti(S2).mq(D).contents(M2) & neti(D).recv_seq(S2) <= net_msg_seq_num(M2)
       & net_msg_mtype(M1) = request_t & net_msg_mtype(M2) = request_t 
       & (S1 ~= S2 | net_msg_seq_num(M1) ~= net_msg_seq_num(M2))
       -> net_msg_req(M1) ~= net_msg_req(M2)

    # A given delegation cannot occur twice in the network

    conjecture neti(S1).mq(D).contents(M1) & neti(D).recv_seq(S1) <= net_msg_seq_num(M1)
       & neti(S2).mq(D).contents(M2) & neti(D).recv_seq(S2) <= net_msg_seq_num(M2)
       & net_msg_mtype(M1) = delegate_t & net_msg_mtype(M2) = delegate_t 
       & (S1 ~= S2 | net_msg_seq_num(M1) ~= net_msg_seq_num(M2))
       -> net_msg_shard(M1) ~= net_msg_shard(M2)

    # The sending seq number is greater than any queue entry

    conjecture mq(D).contents(M) -> ~(send_seq(D) <= net_msg_seq_num(M))

    # No two messages in a queue have the same sequence number

    conjecture mq(D).contents(M1) & mq(D).contents(M2) & M1 ~= M2
        -> net_msg_seq_num(M1) ~= net_msg_seq_num(M2)

    # A sent non-ack message must match any message queue entry with the same
    # sequence number

    conjecture low.sent(M) & net_msg_src(M) = me & net_msg_dst(M) = D
       & mq(D).contents(M2) & net_msg_seq_num(M2) = net_msg_seq_num(M)
       & net_msg_mtype(M) ~= ack_t -> M = M2

    # Following added due to counterexamples

    # A sent non-ack message with seq num >= receiver must be in the
    # corresponding queue

    conjecture low.sent(M) & net_msg_src(M) = S & net_msg_dst(M) = D
      & neti(D).recv_seq(S) <= net_msg_seq_num(M) & net_msg_mtype(M) ~= ack_t
      -> neti(S).mq(D).contents(M)

    # If an ack is sent, the receiving seq_num must be greater

    conjecture low.sent(M) & net_msg_src(M) = S & net_msg_dst(M) = D
      & net_msg_mtype(M) = ack_t -> ~(neti(S).recv_seq(D) <= net_msg_seq_num(M))

    # The sending seq number is greater than non-ack sent message

    conjecture low.sent(M) & net_msg_src(M) = me & net_msg_mtype(M) ~= ack_t
                     -> ~(send_seq(net_msg_dst(M)) <= net_msg_seq_num(M))

    # A message in the queue has correct src and dst and is not ack

    conjecture mq(D).contents(M) -> net_msg_src(M) = me & net_msg_dst(M) = D & net_msg_mtype(M) ~= ack_t
}

module lower(net_impl) = {

    relation sent(M:net_msg)
    
    init ~sent(M)

    action send_msg(msg:net_msg) = {
	sent(msg) := true
    }

    action recv_msg(dst:id, msg:net_msg) = {
	## added fact that only deliver to destination
	assert sent(msg) & dst = net_msg_dst(msg)
    }
    execute recv_msg before net_impl.recv_msg
}


################################################################################
#
#  instantiate some components

# make one server for each id
instance servers(X:id) : server_top(reference,X,neti)

# make one top level network
instance net : net_top(reference,servers,neti)

# delegation map spec
instance dma(X:id) : delegation_map_spec
execute dma.set_ before servers.dm.set_
execute dma.get after servers.dm.get

# network implementation
instance neti(X:id) : net_impl(reference,servers,X,low)

# lower network specification
instance low : lower(neti) 

## input/output specification

object io_spec = {

    relation generated(L:ltime)
    init ~generated(L)

    action generate_request(dst:id,rq:req) = {
	assert ~generated(req_ltime(rq));
	assert L = req_ltime(rq)->
               (req_key(rq) = reference.evs(L).key_ &
                req_otype(rq) = reference.evs(L).type_ &
                (req_otype(rq) = write -> req_data(rq) = reference.evs(L).data_));
	generated(req_ltime(rq)) := true
    }
    execute generate_request before servers.generate_request

    action send_reply(src:id, dst:id, d:data, lt:ltime) = {
	assert reference.evs(lt).data_ = d   # for now, just assert data values are correct
    }
    execute send_reply before neti.send_reply
}

################################################################################
#
# export some stuff

export servers.generate_request
export servers.shard
export neti.recv_msg
export neti.timeout



################################################################################
#
#  verify the servers
#
# This says verify all the actions of "servers", keeping "ref" and "net"
# concrete. 

isolate iso_s = servers with reference,net,dma,io_spec

# Verify the network implementation, using its upper and lower specifications

isolate iso_n = neti with net,low,seq_nums,encoder

# Verify the delegation map implementation. Notice here that
# we can focus on just one server called "me", since the operations
# on delegation maps of different servers don't interfere.

isolate iso_d = servers(me).dm with dma(me),keys
