chandy_lamport

Documentation

Index

• type EndSnapshot
  • func (m EndSnapshot) String() string
• type Link
• type LogEvent
  • func (event LogEvent) String() string
• type Logger
  • func NewLogger() *Logger
  • func (log *Logger) NewEpoch()
  • func (log *Logger) PrettyPrint()
  • func (logger *Logger) RecordEvent(server *Server, event interface{})
• type MarkerMessage
  • func (m MarkerMessage) String() string
• type PassTokenEvent
• type Queue
  • func NewQueue() *Queue
  • func (q *Queue) Empty() bool
  • func (q *Queue) Peek() interface{}
  • func (q *Queue) Pop() interface{}
  • func (q *Queue) Push(v interface{})
• type ReceivedMessageEvent
  • func (m ReceivedMessageEvent) String() string
• type SendMessageEvent
• type SentMessageEvent
  • func (m SentMessageEvent) String() string
• type Server
  • func NewServer(id string, tokens int, sim *Simulator) *Server
  • func (server *Server) AddOutboundLink(dest *Server)
  • func (server *Server) HandlePacket(src string, message interface{})
  • func (server *Server) SendToNeighbors(message interface{})
  • func (server *Server) SendTokens(numTokens int, dest string)
  • func (server *Server) StartSnapshot(snapshotId int)
• type Simulator
  • func NewSimulator() *Simulator
  • func (sim *Simulator) AddForwardLink(src string, dest string)
  • func (sim *Simulator) AddServer(id string, tokens int)
  • func (sim *Simulator) CollectSnapshot(snapshotId int) *SnapshotState
  • func (sim *Simulator) GetReceiveTime() int
  • func (sim *Simulator) InjectEvent(event interface{})
  • func (sim *Simulator) NotifySnapshotComplete(serverId string, snapshotId int)
  • func (sim *Simulator) StartSnapshot(serverId string)
  • func (sim *Simulator) Tick()
• type SnapshotEvent
• type SnapshotMessage
• type SnapshotState
• type StartSnapshot
  • func (m StartSnapshot) String() string
• type SyncMap
  • func NewSyncMap() *SyncMap
  • func (m *SyncMap) Delete(key interface{})
  • func (m *SyncMap) Load(key interface{}) (value interface{}, ok bool)
  • func (m *SyncMap) LoadOrStore(key, value interface{}) (interface{}, bool)
  • func (m *SyncMap) Range(f func(key, value interface{}) bool)
  • func (m *SyncMap) Store(key, value interface{})
• type TokenMessage
  • func (m TokenMessage) String() string

Types

type EndSnapshot

type EndSnapshot struct {
	// contains filtered or unexported fields
}

A message that signifies the end of the snapshot process on a particular server. This is used only for debugging that is not sent between servers.

func (EndSnapshot) String

func (m EndSnapshot) String() string

type Link

type Link struct {
	// contains filtered or unexported fields
}

A unidirectional communication channel between two servers Each link contains an event queue (as opposed to a packet queue)

type LogEvent

type LogEvent struct {
	// contains filtered or unexported fields
}

func (LogEvent) String

func (event LogEvent) String() string

type Logger

type Logger struct {
	// contains filtered or unexported fields
}

func NewLogger

func NewLogger() *Logger

func (*Logger) NewEpoch

func (log *Logger) NewEpoch()

func (*Logger) PrettyPrint

func (log *Logger) PrettyPrint()

func (*Logger) RecordEvent

func (logger *Logger) RecordEvent(server *Server, event interface{})

type MarkerMessage

type MarkerMessage struct {
	// contains filtered or unexported fields
}

A message sent from one server to another during the chandy-lamport algorithm. This is expected to be encapsulated within a `sendMessageEvent`.

func (MarkerMessage) String

func (m MarkerMessage) String() string

type PassTokenEvent

type PassTokenEvent struct {
	// contains filtered or unexported fields
}

An event parsed from the .event files that represent the passing of tokens from one server to another

type Queue

type Queue struct {
	// contains filtered or unexported fields
}

Define a queue -- simple implementation over List

func NewQueue

func NewQueue() *Queue

func (*Queue) Empty

func (q *Queue) Empty() bool

func (*Queue) Peek

func (q *Queue) Peek() interface{}

func (*Queue) Pop

func (q *Queue) Pop() interface{}

func (*Queue) Push

func (q *Queue) Push(v interface{})

type ReceivedMessageEvent

type ReceivedMessageEvent struct {
	// contains filtered or unexported fields
}

A message that signifies receiving of a message on a particular server This is used only for debugging that is not sent between servers

func (ReceivedMessageEvent) String

func (m ReceivedMessageEvent) String() string

type SendMessageEvent

type SendMessageEvent struct {
	// contains filtered or unexported fields
}

An event that represents the sending of a message. This is expected to be queued in `link.events`.

type SentMessageEvent

type SentMessageEvent struct {
	// contains filtered or unexported fields
}

A message that signifies sending of a message on a particular server This is used only for debugging that is not sent between servers

func (SentMessageEvent) String

func (m SentMessageEvent) String() string

type Server

type Server struct {
	Id     string
	Tokens int
	// contains filtered or unexported fields
}

The main participant of the distributed snapshot protocol. Servers exchange token messages and marker messages among each other. Token messages represent the transfer of tokens from one server to another. Marker messages represent the progress of the snapshot process. The bulk of the distributed protocol is implemented in `HandlePacket` and `StartSnapshot`.

func NewServer

func NewServer(id string, tokens int, sim *Simulator) *Server

func (*Server) AddOutboundLink

func (server *Server) AddOutboundLink(dest *Server)

Add a unidirectional link to the destination server

func (*Server) HandlePacket

func (server *Server) HandlePacket(src string, message interface{})

Callback for when a message is received on this server. When the snapshot algorithm completes on this server, this function should notify the simulator by calling `sim.NotifySnapshotComplete`.

func (*Server) SendToNeighbors

func (server *Server) SendToNeighbors(message interface{})

Send a message on all of the server's outbound links

func (*Server) SendTokens

func (server *Server) SendTokens(numTokens int, dest string)

Send a number of tokens to a neighbor attached to this server

func (*Server) StartSnapshot

func (server *Server) StartSnapshot(snapshotId int)

Start the chandy-lamport snapshot algorithm on this server. This should be called only once per server.

type Simulator

type Simulator struct {
	// contains filtered or unexported fields
}

Simulator is the entry point to the distributed snapshot application.

It is a discrete time simulator, i.e. events that happen at time t + 1 come strictly after events that happen at time t. At each time step, the simulator examines messages queued up across all the links in the system and decides which ones to deliver to the destination.

The simulator is responsible for starting the snapshot process, inducing servers to pass tokens to each other, and collecting the snapshot state after the process has terminated.

func NewSimulator

func NewSimulator() *Simulator

func (*Simulator) AddForwardLink

func (sim *Simulator) AddForwardLink(src string, dest string)

Add a unidirectional link between two servers

func (*Simulator) AddServer

func (sim *Simulator) AddServer(id string, tokens int)

Add a server to this simulator with the specified number of starting tokens

func (*Simulator) CollectSnapshot

func (sim *Simulator) CollectSnapshot(snapshotId int) *SnapshotState

Collect and merge snapshot state from all the servers. This function blocks until the snapshot process has completed on all servers.

func (*Simulator) GetReceiveTime

func (sim *Simulator) GetReceiveTime() int

Return the receive time of a message after adding a random delay. Note: since we only deliver one message to a given server at each time step, the message may be received *after* the time step returned in this function.

func (*Simulator) InjectEvent

func (sim *Simulator) InjectEvent(event interface{})

Run an event in the system

func (*Simulator) NotifySnapshotComplete

func (sim *Simulator) NotifySnapshotComplete(serverId string, snapshotId int)

Callback for servers to notify the simulator that the snapshot process has completed on a particular server

func (*Simulator) StartSnapshot

func (sim *Simulator) StartSnapshot(serverId string)

Start a new snapshot process at the specified server

func (*Simulator) Tick

func (sim *Simulator) Tick()

Advance the simulator time forward by one step, handling all send message events that expire at the new time step, if any.

type SnapshotEvent

type SnapshotEvent struct {
	// contains filtered or unexported fields
}

An event parsed from the .event files that represent the initiation of the chandy-lamport snapshot algorithm

type SnapshotMessage

type SnapshotMessage struct {
	// contains filtered or unexported fields
}

A message recorded during the snapshot process

type SnapshotState

type SnapshotState struct {
	// contains filtered or unexported fields
}

State recorded during the snapshot process

type StartSnapshot

type StartSnapshot struct {
	// contains filtered or unexported fields
}

A message that signifies the beginning of the snapshot process on a particular server. This is used only for debugging that is not sent between servers.

func (StartSnapshot) String

func (m StartSnapshot) String() string

type SyncMap

type SyncMap struct {
	// contains filtered or unexported fields
}

An implementation of a map that synchronizes read and write accesses. Note: This class intentionally adopts the interface of `sync.Map`, which is introduced in Go 1.9+ but not available before that. This provides a simplified version of the same class without requiring the user to upgrade their Go installation.

func NewSyncMap

func NewSyncMap() *SyncMap

func (*SyncMap) Delete

func (m *SyncMap) Delete(key interface{})

func (*SyncMap) Load

func (m *SyncMap) Load(key interface{}) (value interface{}, ok bool)

func (*SyncMap) LoadOrStore

func (m *SyncMap) LoadOrStore(key, value interface{}) (interface{}, bool)

func (*SyncMap) Range

func (m *SyncMap) Range(f func(key, value interface{}) bool)

func (*SyncMap) Store

func (m *SyncMap) Store(key, value interface{})

type TokenMessage

type TokenMessage struct {
	// contains filtered or unexported fields
}

A message sent from one server to another for token passing. This is expected to be encapsulated within a `sendMessageEvent`.

func (TokenMessage) String

func (m TokenMessage) String() string