TeSSLa Petri Net Library

Petri nets are a comfortable and widespread formalism for modelling of concurrent systems. The TeSSLa Petri net library enables a comfortable specification and monitoring of Petri net systems.


Basic concept

The TeSSLa Petri net library can be used for a comfortable description of a Petri net structure and its monitoring. Therefore it is necessary to include the library in the main TeSSLa specification.

Since TeSSLa version 2.1.0 the library can be included directly with the include lib "petri_net:x.x.x" statement at the top of the specification where x.x.x indicates the library version. If the library is not yet available in the working directory it is downloaded automatically. For earlier TeSSLa versions the library can be retrieved from the TeSSLa library repository.

For the modelling of the network, the library offers functions in the module PetriNetStructure. Therefore the constant emptyNet and the functions addPlace and addTransitions can be used. For an example consider the following (section of a) Petri net.

Graphical representation of a Petri net

With the TeSSLa Petri net library it can be modeled in the following way:

include lib "petri_net:1.0.0"

import PetriNetStructure

in t0: Events[Int]
in t1: Events[Int]
in t2: Events[Int]
in t3: Events[Int]

def petriNet =
addPlace("Worker", 2, Some(10), true,
addPlace("Todo", 0, None, false,
addPlace("Finished", 0, None, true,
addTransition("t0", Set.empty, Set.singleton(("Worker", 1)), t0,
addTransition("t1", Set.empty, Set.singleton(("Todo", 1)), t1,
addTransition("t2", Set.add(Set.singleton(("Todo", 1)), ("Worker", 2)), Set.add(Set.singleton(("Finished", 1)), ("Worker", 2)), t2,
addTransition("t3", Set.singleton(("Worker", 1)), Set.empty, t3,
emptyNet)))))))

The addPlace macro adds a place to a network. Thereby the first parameter is a string identifier of the place, the second one is the initial token count, the third parameter the (optional) maximal capacity of a place and the fourth one indicates whether the place internally uses FIFO (true) or LIFO (false) policy for managing the tokens,

Likewise the macro addTransition can be used for adding transitions to the network. Therefore every transition receives two sets of place/weight tuples that indicate where the transitions takes how many tuples and where they give it to. Additionally each transition is linked with an integer stream whose events encode how often the transition is firing at which instants.


Monitoring of Petri nets

The TeSSLa Petri net library also contains functions for monitoring the Petri nets modelled as described above. The module PetriNetMonitoring offers three different kinds of monitors with different capabilities.

Monitoring token numbers

With the TokenCntMonitor it is possible to supervise the current number of tokens in a place of the net and to detect under or overflows of the places. Such a monitor can be generated from the Petri net model with use of the macro toTokenCntMonitor:

import PetriNetMonitoring

def tokenCntMonitor = toTokenCntMonitor(petriNet)

out tokenCntMonitor.tokenCnt("Worker") as tokens
out tokenCntMonitor.overflow("Worker") as overflow

The monitor is a structure which contains the macros tokenCnt, underflow and overflow which are parameterized with a place identifier.

stream t0: bubbles
stream t3: bubbles
stream tokens: bubbles
stream overflow: bubbles
---
0: tokens = 2
0: overflow = false
2: t0 = 2
2: tokens = 4
2: overflow = false
4: t0 = 6
4: t3 = 2
4: tokens = 8
4: overflow = false
7: t0 = 3
7: tokens = 11
7: overflow = true

Monitoring time of tokens in places

The PlaceTimeMonitor keeps track of the timestamps when tokens have been added to specific places. Therefore the FIFO/LIFO behavior of the places in the network is simulated.

The usage is similar to the TokenCntMonitor, it offers the monitoring streams tokenTs which carries a list with the timestamps in the place and avgTime as well as maxTime which hold the smallest and the average of the timestamps and can be used to compute the remaining time of tokens in a place:

import PetriNetMonitoring

def placeTimeMonitor = toPlaceTimeMonitor(petriNet)
def maxTime = placeTimeMonitor.maxTime("Worker")

out maxTime
out time(maxTime) - maxTime as remainTime

stream t0: bubbles
stream t3: bubbles
stream maxTime: bubbles
stream remainTime: bubbles
---
0: maxTime = 0
0: remainTime = 0
2: t0 = 2
2: maxTime = 0
2: remainTime = 2
4: t0 = 6
4: t3 = 2
4: maxTime = 2
4: remainTime = 2
7: t3 = 2
7: maxTime = 4
7: remainTime = 3

Important: The PlaceTimeMonitor assumes that no underflows in the network appear.


Monitoring token flows

With the TokenTrackingMonitor it is possible to track the running time of tokens through the network. Therefore, when generating a monitor, a transition of the network is given as parameter and every token that is generated by this transition gets the current timestamp assigned. Since a Petri net does not specify an order in which a place gets tokens from transitions if they fire in parallel and transitions "melt" together tokens from different places also the timestamps of the tokens in such a token tracking network get merged. Thus every token in the TokenTrackingMonitor carries a smallest and greatest timestamp of when tokens that effected this specific token have passed the the corresponding transition.

The TokenTrackingMonitor offers the functions tokenTimes, minTokenTime and maxTokenTime which indicate the timestamp range of tokens in a place, the minimal and the maximal bound of this range. Therefore they return list of time option intervals or options of timestamps.

import PetriNetMonitoring

def tokenTrackingMonitor = toTokenTrackingMonitor(petriNet, "t1")
def timeOption = tokenTrackingMonitor.maxTokenTime("Finished")
def maxTime = filter(getSome(timeOption), isSome(timeOption))

out maxTime

stream t0: bubbles
stream t1: bubbles
stream t2: bubbles
stream maxTime: bubbles
---
1: t0=6
2: t1 = 2
4: t1 = 1
4: t2 = 1
4: maxTime = 2
7: t2 = 1
7: t1 = 1
7: maxTime = 4
9: t2 = 1
9: maxTime = 7

Important: The TokenTrackingMonitor assumes that no underflows in the network appear.


Connection with other parts of the TeSSLa specification

The TeSSLa ecosystem enables an easy integration of the Petri net monitoring with other TeSSLa specified streams or libraries. E.g. the firing of transitions could be bound to properties specified in past LTL and monitored properties over the network could be formulated in TDDL logic:

include lib "petri_net:1.0.0"
include lib "tddl:1.0.1"

import LTL
import PetriNetStructure
import PetriNetMonitoring
import TDDL

in x: Events[Int]
in y: Events[Int]

def petriNet =
  addPlace("s0", 10, None, false,
  addPlace("s2", 0, None, false,
  addTransition("t0", Set.singleton(("s0", 1)), Set.singleton(("s1", 1)), if since(x == 3, y != 5) then 1 else 0,
  emptyNet)))

def tokenCntMonitor = toTokenCntMonitor(petriNet)

def s1TokenCnt = tokenCntMonitor.tokenCnt("s1")
def tddlMonitor = Monitor(O(3, (Some(2), Some(20))))

def mon = tddlMonitor(s1TokenCnt)


Remarks

Possible extensions to the library would be the support of colored tokens and an extension with timed transitions. The latter one can be done quite easily by another layer around the PN structure which introduces additional "hidden" places that hold back tokens after firing of a transition and give them to the original place triggered by a delayed transition.

If you want to contribute to the development of the Petri net library please contact the TeSSLa development team.