edentv-4.9.1: RTSEventsParser.hs
{- The Eden Trace Viewer (or simply EdenTV) is a tool that can generate diagrams
to visualize the behaviour of Eden programs.
Copyright (C) 2005-2012 Philipps Universitaet Marburg
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software Foundation,
Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
-}
{-# OPTIONS -XNamedFieldPuns #-}
{-# LANGUAGE BangPatterns #-}
module RTSEventsParser (
traceRTSFile,
Err(..)
) where
import TinyZipper
import qualified EdenTvType as E
import GHC.RTS.Events
import qualified Control.Exception as C
import Control.Monad.Error (runErrorT)
import Data.Binary.Get
import Data.ByteString.Lazy (ByteString)
import Data.Either
import Data.Maybe
import Data.Tree
import System.FilePath (takeExtension)
import qualified Data.ByteString.Lazy as BS
import qualified Data.Map as M
import qualified Data.Sequence as S
import Debug.Trace
type RTSEvents = (E.MachineID, [Event])
data Err a = Ok !a | Failed String
type Lookuptable = [(E.MachineID, [(Int, Int)])]
----------------------------------
--
-- Reader function: parseZipFile
-- Reads a zip File containing binary eventlog files
-- to a List of RTSEvents
--
----------------------------------
-- We need to catch Exceptions here, because getEventLog (GHC.RTS.Events) uses Lazy Bytestrings and
-- non-strict Get!
parseZipFile :: FilePath -> IO (Either String [RTSEvents])
parseZipFile zipfile = C.catch (readAll zipfile) catchBadFile
where catchBadFile :: C.SomeException -> IO (Either String [RTSEvents])
catchBadFile e = return (Left ("ParseZipFile: Bad File (due to: "++(show e)++")"))
parseEventlogFile :: FilePath -> IO (Either String [RTSEvents])
parseEventlogFile f = do s <- BS.readFile f
log <- readGhcEventsSingle s
return $ log >>= eventLogToEvents >>= (return . (:[]))
-- similar to readEventLogFromFile, but uses a ByteString
-- parses a binary *.eventlog file
readGhcEventsSingle :: ByteString -> IO (Either String EventLog)
readGhcEventsSingle s = return $ runGet (do v <- runErrorT $ getEventLog
m <- isEmpty
m `seq` return v) $ s
-- convert an EventLog to a List of its Events, paired with
-- the machine number
eventLogToEvents :: EventLog -> Either String RTSEvents
eventLogToEvents evtlog =
let ghcevents = map ce_event $ sortEvents (events $ dat evtlog)
mID = getMachineID ghcevents
in either (\err -> Left err) (\id -> Right (id, ghcevents)) mID
where getMachineID :: [Event] -> Either String E.MachineID
getMachineID ((Event{time, spec=CreateMachine{machine}}):_) =
Right $ fromIntegral machine
getMachineID (_:xs) = getMachineID xs
--TH: removed error with a default machine 1
-- machineid 0 might be better, but 1 is assumed to be the main machine
-- machines only exist in Eden (parevents) traces
--getMachineID [] = Left "eventLogToEvents: No CreateMachine Event!"
getMachineID [] = Right 1
-- now we can read all files contained in a zip archive
readAll :: FilePath -> IO (Either String [RTSEvents])
readAll zippath = do
files' <- readZip zippath
either (\err -> return $ Left err)
(\files -> do
eventlogs <- sequence $ map readGhcEventsSingle files
let (errparse, logs) = partitionEithers eventlogs
case null errparse of
True -> do
let (errevts, events) =
partitionEithers $ map eventLogToEvents logs
case null errevts of
False -> return $ Left ("readAll: Events error "++(show errevts))
True -> return $ Right events
False -> return $ Left "readAll: Parse error")
files'
---------------------------------------------------------------
-- parseAdditional
-- parses Additional information (needed for, or not covered by
-- the old legacy parser) from RTSEvents
---------------------------------------------------------------
data AddInfo = AddInfo {
ai_threadTable :: !ThreadTable, -- get Proc# from Thread#
ai_mainTime :: !E.Seconds, -- realtime of 1st machine
ai_syncTable :: ![(E.MachineID, E.Seconds)], -- offsets to mainTime
ai_startupDiff :: !(E.Seconds, [(E.MachineID, E.Seconds)]), -- offsets to Startup
ai_endTable :: ![(E.MachineID, E.Seconds)], -- end Times of Machines
ai_maxEndTime :: !E.Seconds, -- max End Time
ai_minCMTime :: !E.Seconds, -- earliest CreateMachine Time
ai_maxSUDiff :: !E.Seconds, -- max startup offset
ai_rcvLengths :: ![((E.MachineID,[(Int,E.Seconds)]), E.Seconds, E.Seconds)], -- used for rect Blocks in Grouped Messages
ai_leftMsgs :: ![E.Message] -- Messages which are left out by old ETV
}
type ThreadTable = [(E.MachineID, [(Int, Int)])]
parseAdditional :: Bool -> [RTSEvents] -> AddInfo
parseAdditional ignoreMessages = mergeInfo .
map (\(mID, evts) -> (parseAdditionalSingle evts (emptyInfo mID)) ignoreMessages)
mergeInfo :: [AddInfoSingle] -> AddInfo
mergeInfo aiss =
let threadTable = map (\ais -> let mID = ais_machineID ais
tTable = ais_threadTable ais
in mID `seq` (length tTable) `seq` (mID, tTable)) aiss
mainTime = ais_realtime $ head $ filter (\ais -> ais_machineID ais == 1) aiss
syncTable = map (\ais -> let mID = ais_machineID ais
realT = ais_realtime ais
in mID `seq` realT `seq` (mID, realT - mainTime)) aiss
cmTimes = map (\ais -> let mID = ais_machineID ais
cmT = convertTimestampWithTable syncTable mID (ais_CMTimestamp ais)
in mID `seq` cmT `seq` (mID, cmT)
) aiss
minCMTime = minimum $ map snd cmTimes
startupTimes = map (\ais -> let mID = ais_machineID ais
suT = convertTimestampWithTable syncTable mID (ais_SUTimestamp ais)
in mID `seq` suT `seq` (mID, suT)
) aiss
maxStartupTime = maximum $ map snd startupTimes
startupDiff = (maxStartupTime, map (\(mID, s) -> (mID, maxStartupTime - s)) startupTimes)
maxStartupDiff = maximum $ zipWith (\(_, diffs) (_,end) -> if diffs + end > maxEndTime then (diffs+end)-maxEndTime else 0) (snd startupDiff) endTimes
endTimes = map (\ais -> let mID = ais_machineID ais
leT = convertTimestampWithTable syncTable mID (ais_LETimestamp ais)
in mID `seq` leT `seq` (mID, leT)
) aiss
maxEndTime = maximum $ map snd endTimes
rcvLengths = concatMap (\ais -> let mID = ais_machineID ais
convert = convertTimestampWithTable syncTable mID
in map (\(pids, ts, te) -> let pidM = reverse $ map (\(pid, t) -> (pid, convert t)) pids
cts = convert ts
cte = convert te
in mID `seq` (length pidM) `seq` cts `seq` cte `seq` ((mID, pidM), cts, cte)
) (ais_receiveLengths ais)) aiss
leftMsgEvents :: [(E.MachineID, Event)]
leftMsgEvents = concatMap (\ais -> map (\e -> (ais_machineID ais, e)) (ais_leftMessages ais)) aiss
leftMsgs :: (E.OpenMessagesPerProcess, [E.Message])
leftMsgs = foldr processLeftMsgs (M.empty, []) leftMsgEvents
left :: [E.Message]
left = snd leftMsgs
processLeftMsgs :: (E.MachineID, Event) -> (E.OpenMessagesPerProcess, [E.Message])
-> (E.OpenMessagesPerProcess, [E.Message])
processLeftMsgs (mID, Event{time, spec}) oe
= let convertTimestamp = convertTimestampWithSync (toSyncTime syncTable mID)
in case spec of
SendMessage{
mesTag,
senderProcess,
senderThread,
receiverMachine,
receiverProcess,
receiverInport}
-> insertLeftOutMessages (E.UserProcess mID (fromIntegral senderProcess))
(E.OSM (convertTimestamp time)
(E.UserProcess (fromIntegral receiverMachine) (fromIntegral receiverProcess))
(fromIntegral senderThread)
(fromIntegral receiverInport)
(readTag mesTag)) oe
ReceiveMessage{
mesTag,
receiverProcess,
receiverInport,
senderMachine,
senderProcess,
senderThread,
messageSize}
-> insertLeftOutMessages (E.UserProcess (fromIntegral senderMachine) (fromIntegral senderProcess))
(E.ORM (convertTimestamp time)
(E.UserProcess mID (fromIntegral receiverProcess))
(fromIntegral senderThread)
(fromIntegral receiverInport)
(readTag mesTag)
(fromIntegral messageSize)) oe
SendReceiveLocalMessage{
mesTag,
senderProcess,
senderThread,
receiverProcess,
receiverInport}
-> insertLeftOutLocalMessage (E.MSG (E.UserProcess mID (fromIntegral senderProcess),
fromIntegral senderThread,
(E.UserProcess mID (fromIntegral receiverProcess)),
fromIntegral receiverInport)
(convertTimestamp time)
(convertTimestamp time)
(readLocalTag mesTag)
0
) oe
_ -> oe
evaluateMessages [] = []
evaluateMessages ((E.MSG _ _ _ _ _):msgs) = evaluateMessages msgs
in (length threadTable) `seq`
(length rcvLengths) `seq`
(evaluateMessages left) `seq`
AddInfo { ai_threadTable = threadTable,
ai_mainTime = mainTime,
ai_syncTable = syncTable,
ai_startupDiff = startupDiff,
ai_endTable = endTimes,
ai_maxEndTime = maxEndTime,
ai_rcvLengths = rcvLengths,
ai_minCMTime = minCMTime,
ai_maxSUDiff = maxStartupDiff,
ai_leftMsgs = left
}
emptyInfo :: E.MachineID -> AddInfoSingle
emptyInfo mID = AddInfoSingle { ais_machineID = mID,
ais_threadTable = [],
ais_CMTimestamp = 0,
ais_SUTimestamp = 0,
ais_LETimestamp = 0,
ais_realtime = 0,
ais_receiveLengths = [],
ais_leftMessages = [],
ais_tCollect = False,
ais_tStart = 0,
ais_tPIDS = []
}
data AddInfoSingle = AddInfoSingle {
ais_machineID :: !E.MachineID,
ais_threadTable :: ![(Int,Int)], -- (Thread#, Proc#)
ais_CMTimestamp :: !Timestamp, -- Timestamp of CreateMachine
ais_SUTimestamp :: !Timestamp, -- Timestamp of Startup
ais_LETimestamp :: !Timestamp, --- Timestamp of last Event
ais_realtime :: !E.Seconds,
ais_receiveLengths :: ![([(Int, Timestamp)], Timestamp, Timestamp)],
ais_leftMessages :: ![Event],
ais_tCollect :: !Bool,
ais_tStart :: !Timestamp,
ais_tPIDS :: ![(Int,Timestamp)]
}
parseAdditionalSingle :: [Event] -> AddInfoSingle -> Bool -> AddInfoSingle
parseAdditionalSingle [] aux _ = aux
parseAdditionalSingle ((Event{time,
spec=AssignThreadToProcess{thread,process}}):evts) aux ignoreMessages =
let oldTT = ais_threadTable aux
newTT@((!t, !p): (!oldTT')) = (fromIntegral thread, fromIntegral process):oldTT
in parseAdditionalSingle evts (aux{ais_threadTable = newTT,ais_LETimestamp= time}) ignoreMessages
parseAdditionalSingle ((Event{time,
spec=CreateMachine{realtime}}):evts) aux ignoreMessages =
let newRealtime = ((fromIntegral realtime)/10e7)-((fromIntegral time) / 10e8)
in parseAdditionalSingle evts (aux{ais_CMTimestamp = time, ais_realtime = newRealtime,ais_LETimestamp= time}) ignoreMessages
parseAdditionalSingle ((Event{time,
spec=Startup{}}):evts) aux ignoreMessages = parseAdditionalSingle evts (aux{ais_SUTimestamp = time, ais_LETimestamp= time}) ignoreMessages
parseAdditionalSingle (e@(Event{time, spec=SendMessage{}}):evts) aux@(AddInfoSingle{ais_leftMessages=oldMsg}) ignoreMessages
| ignoreMessages = parseAdditionalSingle evts aux ignoreMessages
| otherwise = parseAdditionalSingle evts (aux{ais_leftMessages = (e:oldMsg), ais_LETimestamp= time}) ignoreMessages
parseAdditionalSingle ((Event{time,
spec=EdenStartReceive}):evts) aux ignoreMessages =
parseAdditionalSingle evts (aux{ais_tCollect = True,ais_tStart=time,ais_LETimestamp= time }) ignoreMessages
parseAdditionalSingle ((Event{time,
spec=EdenEndReceive}):evts) aux@(AddInfoSingle{ais_receiveLengths=oldRL, ais_tStart, ais_tPIDS}) ignoreMessages =
ais_tStart `seq` ais_tPIDS `seq` oldRL `seq` parseAdditionalSingle evts (aux{ais_tCollect = False,ais_receiveLengths=((ais_tPIDS, ais_tStart, time):oldRL), ais_tPIDS=[], ais_LETimestamp= time}) ignoreMessages
parseAdditionalSingle ((Event{time,
spec=CreateProcess{process}}):evts) aux@(AddInfoSingle{ais_tCollect=True, ais_tPIDS=oldPIDs}) ignoreMessages =
let p = fromIntegral process
in p `seq` oldPIDs `seq` parseAdditionalSingle evts (aux{ais_tPIDS=((p,time):oldPIDs),ais_LETimestamp= time}) ignoreMessages
parseAdditionalSingle (e@(Event{time, spec=ReceiveMessage{receiverProcess,mesTag}}):evts) aux@(AddInfoSingle{ais_tCollect=b, ais_tPIDS=oldPIDs,ais_leftMessages=oldMsg}) ignoreMessages
| ignoreMessages = parseAdditionalSingle evts aux ignoreMessages
| b = case mesTag of RFork -> parseAdditionalSingle evts (aux{ais_leftMessages = (e:oldMsg),ais_LETimestamp=time}) ignoreMessages
_ -> let rp = fromIntegral receiverProcess
in rp ` seq` parseAdditionalSingle evts (aux{ais_leftMessages = (e:oldMsg),ais_tPIDS=((fromIntegral receiverProcess,time):oldPIDs),ais_LETimestamp=time}) ignoreMessages
| otherwise = parseAdditionalSingle evts (aux{ais_leftMessages = (e:oldMsg),ais_LETimestamp=time}) ignoreMessages
parseAdditionalSingle (e@(Event{time, spec=SendReceiveLocalMessage{}}):evts) aux@(AddInfoSingle{ais_leftMessages=oldMsg}) ignoreMessages
| ignoreMessages = parseAdditionalSingle evts aux ignoreMessages
| otherwise = parseAdditionalSingle evts (aux{ais_leftMessages = (e:oldMsg), ais_LETimestamp= time}) ignoreMessages
parseAdditionalSingle (e:evts) aux ignoreMessages = parseAdditionalSingle evts (aux{ais_LETimestamp= time e}) ignoreMessages
-- This function is called for every SendMessage or ReceiveMessage that
-- was found in the additional events. It takes care of grouping corresponding
-- send and receive message events into a single message.
--
-- copy of legacy Code, used to just extract Head Messages
insertLeftOutMessages :: E.ProcessID -- sender process id
-> E.OpenMessageEvent -- the message event to process
-> (E.OpenMessagesPerProcess, -- messages that could not yet be matched with their
-- corresponding close/open message grouped per process
[E.Message]) -- complete messages for which the send and
-- and receive event could be matched
-> (E.OpenMessagesPerProcess, [E.Message])
insertLeftOutMessages senderProcessId newMessage ocMsgs@(openMessages, closedMessages)
| or [(tag == E.Head), (tag == E.DataMes)]
= if (min (E.pId senderProcessId) (E.pId receiveProcessId)) >= 0
then openMessages' `seq` closedMessages' `seq` (openMessages', closedMessages')
else ocMsgs
| otherwise = ocMsgs
where (receiveProcessId, tag) = case newMessage of
E.ORM _ p _ _ r _ -> (p, r)
E.OSM _ p _ _ r -> (p, r)
(openMessages', closedMessages') = searchID senderProcessId newMessage openMessages closedMessages
insertLeftOutLocalMessage :: E.Message -> (E.OpenMessagesPerProcess, [E.Message]) -> (E.OpenMessagesPerProcess, [E.Message])
insertLeftOutLocalMessage m (oms, msgs) = let msgs' = m : msgs
in msgs' ` seq` (oms, msgs')
-- This function receives a process id, a message event (send or receive) for a message
-- that was sent from this process, a map of open messages for all processes
-- and a list of closed messages.
--
-- If there already are message events for the given process, the corresponding
-- send/receive message is searched in the open messages of the process.
-- If a message can be 'closed' (a SendMessage can be found for a ReceiveMessage,
-- or a ReceiveMessage can be found for a SendMessage), the merged message is
-- added to the list of closed messages.
--
searchID :: E.ProcessID -> E.OpenMessageEvent -> E.OpenMessagesPerProcess -> [E.Message]
-> (E.OpenMessagesPerProcess, [E.Message])
searchID senderProcess newOpenMessage openMessagesPerProcess closedMessages
= trySearchProcess (M.lookup senderProcess openMessagesPerProcess)
where
-- the key to store the message event in a map
messageKey = getKeyForOpenMessage newOpenMessage
-- the key a corresponding 'closing' message event would have
closingMessageKey = getClosingMessageKey newOpenMessage
-- a compressed representation of the message which is used as
-- value to store the message in a map (all other information
-- about the message is already contained in the key)
smallNewOpenMessage = getSmallOpenMessage newOpenMessage
trySearchProcess :: Maybe E.OpenMessages -> (E.OpenMessagesPerProcess, [E.Message])
-- this is the first message of the process, create a new entry in the map
trySearchProcess (Nothing) = (openMessagesPerProcess', closedMessages)
where
openMessagesPerProcess' = M.insert senderProcess processEntry openMessagesPerProcess
processEntry = M.singleton messageKey (S.singleton smallNewOpenMessage)
-- there is already an entry for the process, look for an closing message event
trySearchProcess (Just openMessages)
= trySearchClosingMessage (M.lookup closingMessageKey openMessages) openMessages
trySearchClosingMessage :: Maybe (S.Seq E.SmallOpenMessageEvent) -> E.OpenMessages
-> (E.OpenMessagesPerProcess, [E.Message])
-- there is no matching closing message event yet, so store the event
trySearchClosingMessage (Nothing) openMessages
= (openMessagesPerProcess', closedMessages)
where
-- update the process entry in the map
openMessagesPerProcess' :: E.OpenMessagesPerProcess
openMessagesPerProcess' = M.insert senderProcess openMessages' openMessagesPerProcess
-- add the event to the map of open events for this process:
-- if there already are open events for this process that use the same
-- channel (same type/rcvProcess/inport/outport/type), thus use the same
-- map key, then enqueue the new message to the sequence of these messages.
openMessages' :: E.OpenMessages
openMessages' = M.insertWith enqueueEvent messageKey (S.singleton smallNewOpenMessage) openMessages
-- this functions takes care of appending the new event to the sequence
-- of already existing events with the same map key
enqueueEvent :: S.Seq E.SmallOpenMessageEvent -- sequence which contains only the new event
-> S.Seq E.SmallOpenMessageEvent -- old events with the same key
-> S.Seq E.SmallOpenMessageEvent
enqueueEvent newOpenMessageSeq oldOpenMessages = enqueue oldOpenMessages newOpenMessage
where
newOpenMessage = S.index newOpenMessageSeq 0
enqueue = (S.|>)
-- if there is a corresponding event, 'close' the message
trySearchClosingMessage (Just oldMessages) openMessages
= (openMessagesPerProcess', closedMessages')
where
openMessagesPerProcess' :: E.OpenMessagesPerProcess
openMessagesPerProcess' =
-- if this was the last message event of the process,
-- remove the process' map entry
if (M.null openMessages')
then M.delete senderProcess openMessagesPerProcess
else M.insert senderProcess openMessages' openMessagesPerProcess
openMessages' :: E.OpenMessages
openMessages' =
-- if there are no more events with the same key as the
-- closing event, remove the key from the map
if ((S.length oldMessages) == 1)
then M.delete closingMessageKey openMessages
else M.insert closingMessageKey (S.drop 1 oldMessages) openMessages
closedMessages' :: [E.Message]
closedMessages' = closedMessage : closedMessages
closedMessage :: E.Message
closedMessage = closeMessage newOpenMessage oldMessage
oldMessage :: E.SmallOpenMessageEvent
oldMessage = S.index oldMessages 0
-- Merges matching send- and receive-message events into a single message
closeMessage :: E.OpenMessageEvent -> E.SmallOpenMessageEvent -> E.Message
closeMessage (E.OSM sendTime rcvProcess outport inport reason) (E.SmallORM rcvTime size)
= E.MSG (senderProcess, outport, rcvProcess, inport) sendTime rcvTime reason size
closeMessage (E.ORM rcvTime rcvProcess outport inport reason size) (E.SmallOSM sendTime)
= E.MSG (senderProcess, outport, rcvProcess, inport) sendTime rcvTime reason size
-- Creates a 'small' representation of a message event, which is
-- used to store the message in a map.
getSmallOpenMessage :: E.OpenMessageEvent -> E.SmallOpenMessageEvent
getSmallOpenMessage (E.OSM sendTime rcvProcess outport inport reason)
= E.SmallOSM sendTime
getSmallOpenMessage (E.ORM sendTime rcvProcess outport inport reason size)
= E.SmallORM sendTime size
-- Returns a map key to retrieve a message event which would close
-- the given message event.
getClosingMessageKey :: E.OpenMessageEvent -> E.OpenMessageKey
getClosingMessageKey (E.ORM _ rcvProcess outport inport reason _)
= getKeyForOpenMessage (E.OSM __ rcvProcess outport inport reason)
getClosingMessageKey (E.OSM _ rcvProcess outport inport reason)
= getKeyForOpenMessage (E.ORM __ rcvProcess outport inport reason ___)
__ = 0 / 0
___ = 0
-- Returns a key which is used store a message event in a map.
getKeyForOpenMessage :: E.OpenMessageEvent -> E.OpenMessageKey
getKeyForOpenMessage (E.ORM _ rcvProcess outport inport reason _)
= (E.TORM, rcvProcess, outport, inport, reason)
getKeyForOpenMessage (E.OSM _ rcvProcess outport inport reason)
= (E.TOSM, rcvProcess, outport, inport, reason)
--------------------------------------------------------------------------------
-- Main Function
--------------------------------------------------------------------------------
traceRTSFile :: FilePath -> Bool -> IO (Err E.Events)
traceRTSFile file ignoreMessages =
do ghcevents' <- if isPareventsFile file
then parseZipFile file
else parseEventlogFile file
case ghcevents' of
Left err -> return (Failed err)
Right ghcevents ->
do return (Ok finishedEvents)
where processedEvents = closeOpenLists $ process ghcevents emptyoe
additionalInfo = parseAdditional ignoreMessages ghcevents
finishedEvents = if ignoreMessages
then finish processedEvents
else injectProcMessages $ finish processedEvents
-- finish ((lom,lop,lot),mst, sudiff, (ml,aml,hml,pt, rcvl ), (min_t, m_t,msudiff,mmsg,mld),nums) =
-- ((lom,lop,lot),mst, ai_startupDiff additionalInfo, (ml,ai_leftMsgs additionalInfo,hml,pt, ai_rcvLengths additionalInfo), (min_t, m_t,ai_maxSUDiff additionalInfo,mmsg,mld),nums)
finish events = let !(maxSU, startupOffs) = ai_startupDiff additionalInfo
!(ml, aml, hml, pt, rcvl) = E.messagelist events
in events { E.maxStartup = maxSU,
E.startupOffsets = startupOffs,
E.messagelist = (ml, ai_leftMsgs additionalInfo, hml, pt, ai_rcvLengths additionalInfo),
E.max_t_diff = ai_maxSUDiff additionalInfo
}
{-
-- START HACK
-- TODO needs rewrite
-}
-- Manually build Process Table
-- and ProcMessages (needed because wrong results
-- when those messages are processed in a 'wrong' order, and
-- finding the right order is just as complex as this)
sendRForks = map (\(mID, evts) -> (mID, filter (isSendRFork) evts)) ghcevents
isSendRFork :: Event -> Bool
isSendRFork evt = case spec evt of
SendMessage{mesTag=RFork} -> True
_ -> False
rcv_crtRForks = map (\(mID, evts) -> (mID, filter (\evt -> case spec evt of ReceiveMessage{mesTag=RFork} -> True; CreateProcess{} -> True; _ -> False) evts)) ghcevents
injectProcMessages :: E.Events -> E.Events
--injectProcMessages ((lom,lop,lot),mst, inj ,(ml,aml,hml,pt, rcvl ),stats,nums) = ((lom,lop,lot),mst, inj,(ml++procMessages,aml,hml,processTree, rcvl ),stats,nums)
injectProcMessages events = let (ml, aml, hml, pt, rcvl) = E.messagelist events
in events { E.messagelist = (ml++procMessages,aml,hml,processTree, rcvl) }
childTable :: [(E.ProcessID, [E.ProcessID])]
procMessages :: [E.Message]
(childTable,procMessages) = createChildTable sendRForks rcv_crtRForks [] []
processTree :: E.ProcessTree
processTree = unfoldTree buildTree (E.UserProcess 1 1, fromJust (lookup (E.UserProcess 1 1) childTable))
buildTree :: (E.ProcessID, [E.ProcessID]) -> (E.ProcessID, [(E.ProcessID, [E.ProcessID])])
buildTree (pid, pids) = (pid, map (\p -> let mps = lookup p childTable
in case mps of
Nothing -> (p, [])
Just ps -> (p, ps)) pids)
createChildTable :: [(E.MachineID, [Event])] -> [(E.MachineID, [Event])] -> [(E.ProcessID, [E.ProcessID])] -> [E.Message] -> ([(E.ProcessID, [E.ProcessID])], [E.Message])
createChildTable ((mID, evt:evts):machines) rcvers table procMsgs=
let SendMessage{senderProcess,receiverMachine} = spec evt
senderProc = E.UserProcess mID (fromIntegral sProc)
receiverMach = fromIntegral receiverMachine
sMach = fromIntegral mID
sProc = senderProcess
sTime = convertTimestampWithSync (toSyncTime (ai_syncTable additionalInfo) mID) (time evt)
Just rcvEvents = lookup receiverMach rcvers
(childId, remainingEvents, child) = nextChildId rcvEvents
childProcess = E.UserProcess receiverMach childId
newRcvers = updateReceivers receiverMach rcvers remainingEvents
newTable = addChild table senderProc childProcess
newProcMsgs = child:procMsgs
addChild :: [(E.ProcessID, [E.ProcessID])] -> E.ProcessID -> E.ProcessID -> [(E.ProcessID, [E.ProcessID])]
addChild (cur@(pid, childs):pids) father child
| pid == father = ((pid, (child:childs)):pids)
| otherwise = cur : (addChild pids father child)
addChild [] father child = [(father, [child])]
updateReceivers :: E.MachineID -> [(E.MachineID, [Event])] -> [Event] -> [(E.MachineID, [Event])]
updateReceivers i (mach@(mID, evts):machines) update
| mID == i = ((mID, update):machines)
| otherwise = mach : (updateReceivers i machines update)
updateReceivers _ [] _ = error "not updated"
nextChildId (e:es) = case spec e of
ReceiveMessage{senderMachine,senderProcess,
messageSize} -> if senderMachine == sMach && senderProcess == sProc
then let (Event{spec=CreateProcess{process}}, rest) = getNextAndRemove (\evt -> case spec evt of CreateProcess{} -> True; _ -> False) es
childId = fromIntegral process
channelID = (senderProc, 0, E.UserProcess receiverMach childId, 0)
rTime = convertTimestampWithSync (toSyncTime (ai_syncTable additionalInfo) receiverMach) (time e)
child = E.MSG channelID sTime rTime E.RFork (fromIntegral messageSize)
in (childId, rest, child)
else let (id, rest, child) = nextChildId es
in (id, e:rest, child)
_ -> let (id, rest, child) = nextChildId es
in (id, e:rest, child)
nextChildId [] = error "next child not found"
getNextAndRemove :: (a -> Bool) -> [a] -> (a, [a])
getNextAndRemove p (x:xs)
| p x = (x,xs)
| otherwise = let (y,ys) = getNextAndRemove p xs
in (y,x:ys)
getNextAndRemove _ [] = (error "next not found", [])
in createChildTable ((mID, evts):machines) newRcvers newTable newProcMsgs
createChildTable ((mID, []):machines) rcvers table procMsgs = createChildTable machines rcvers table procMsgs
createChildTable [] ecv table procMsgs = (table,procMsgs)
{-
-- END HACK
-}
-- from old haskell EdenTV
-- initial events structure
emptyoe = (([],[],[]),[(0,firstTime)], fml firstTime, (firstTime, firstTime, 0, -1))
where firstTime = ai_minCMTime additionalInfo
-- from old haskell EdenTV
-- creates initial MessageList
fml :: E.Seconds -> E.OpenMessageList
fml t = (M.empty,[],([(1,[(E.UserProcess 1 1,E.OSM t (E.UserProcess 0 0) (-1) (-1) E.RFork)],[(E.UserProcess 0 0,E.ORM t (E.UserProcess 1 1) (-1) (-1) E.RFork 4)],[])],[(E.UserProcess 0 0,[])],Node (E.UserProcess 0 0) []),([],0,[]))
-- walks over the list of machines (which contains all events for this machine) and inserts all events
-- in an accumulated OpenEvents structure
-- when no events are left, we are finished and just need to close open lists in the generated structure
process :: [(E.MachineID, [Event])] -> E.OpenEvents -> E.OpenEvents
process ((mID, (evt:evts)):machines) oe = let convertTimestamp = convertTimestampWithSync (toSyncTime (ai_syncTable additionalInfo) mID)
newOE = insertEvent convertTimestamp (ai_threadTable additionalInfo) (mID,evt) oe ignoreMessages
in seq newOE $ process ((mID,evts):machines) newOE
process ((_,[]):machines) oe = process machines oe
process [] oe = oe
--------------------------------------------------------------------------------
-- Helpers
--------------------------------------------------------------------------------
-- converts a timestamp in TICKS to Seconds with respect
-- to a realtime offset 'sync'
convertTimestampWithSync :: E.Seconds -> Timestamp -> E.Seconds
convertTimestampWithSync sync x = sync + (fromIntegral x / 10e8)
convertTimestampWithTable :: [(E.MachineID, Double)] -> E.MachineID -> Timestamp -> E.Seconds
convertTimestampWithTable syncTable mID timestamp =
let offset = toSyncTime syncTable mID
in convertTimestampWithSync offset timestamp
toSyncTime :: [(E.MachineID, Double)] -> E.MachineID -> E.Seconds
toSyncTime ((mId, t):machines) i
| mId == i = t
| otherwise = toSyncTime machines i
toSyncTime [] _ = 0
-- converts a MessageTag to the ReasonType specified in old legacy format
readTag :: MessageTag -> E.ReasonType
readTag tag = case tag of
RFork -> E.RFork
Connect -> E.Connect
DataMes -> E.DataMes
Head -> E.Head
Constr -> E.Constr
Part -> E.Part
Terminate -> E.Terminate
_ -> E.Default
readLocalTag :: MessageTag -> E.ReasonType
readLocalTag tag = case tag of
Head -> E.LocalHead
DataMes -> E.LocalDataMes
_ -> error "readLocalTag was given a non-local tag"
-- finds the matching Process# to a given MachineId and Thread#
-- inside the ThreadTable
-- returns -1 for unassigned processes
lookupProcess :: ThreadTable -> E.MachineID -> Int -> Int
lookupProcess ((machine, es):ts) sMachine sThread =
if machine == sMachine
then lookupProcess' es sThread
else lookupProcess ts sMachine sThread
where lookupProcess' lst@((tnum,pnum):xs) stnum = if tnum == stnum
then pnum
else lookupProcess' xs stnum
lookupProcess' [] stnum = -1
lookupProcess [] sMachine sThread = -1
isPareventsFile :: FilePath -> Bool
isPareventsFile = (flip elem [".parevents", ".zip"]) . takeExtension
--------------------------------------------------------------------------------
-- Legacy Code
--------------------------------------------------------------------------------
-- legacy code of old haskell EdenTV
-- modified to work with new Tracefile Format
insertEvent :: (Timestamp -> E.Seconds) -> Lookuptable -> (E.MachineID, Event) -> E.OpenEvents -> Bool -> E.OpenEvents
insertEvent convertTimestamp lutable (mId, Event{time, spec})
oldEvents@(oEvts@(ms,ps,ts),mts,ocMsgs,(minTime,maxTime,nP,maxLD)) ignoreMessages
| isMsgEvent && ignoreMessages = oldEvents
| isMsgEvent = newMsgEvents `seq` ((newMs, newPs,ts),mts,newMsgEvents,(minTime,newMax,nP,maxLD)) -- process the event
| isTrdEvent = let (newTData, mPE) = newTrdEvents
(newPData, mME) = case mPE of
[] -> (ps, []) -- no virtual process event
pe -> insPEventList pe ps -- insert generated process event
newMData = case mME of
[] -> ms -- no virtual machine event
me -> insMEventList mId me ms -- insert generated machine event
in seq newMData ((newMData, newPData, newTData),mts,ocMsgs,(minTime,newMax,nP,maxLD))
| isPrcEvent = let (newPData,mME) = newPrcEvents
newMData = case mME of
Nothing -> ms -- no virtual machine event
Just me -> insMEvent mId me ms -- virtual machine event generated
in seq newMData ((newMData,newPData,ts),mts,newOcMsgs,(minTime,newMax,newNP,maxLD))
| isMchEvent = let newData = case gcEvents of
Nothing -> (newMchEvents, ps, ts)
Just (p,t) -> let newPE = insPeByMID mId p ps
newTE = insTeByMID mId t ts
in (newMchEvents, newPE, newTE)
in seq newData (newData,newMchTimes,ocMsgs,(min',max',nP,maxLD'))
| otherwise = oldEvents -- ignore unknown events
where
(isMsgEvent, newMsgEvents, newMs, newPs) = case spec of
SendMessage{mesTag=Connect} -> (False, undefined, undefined, undefined) -- ignore
SendMessage{
mesTag,
senderProcess,
senderThread,
receiverMachine,
receiverProcess,
receiverInport} -> (True, createNewMsgEvents
(E.UserProcess mId (fromIntegral senderProcess))
(E.OSM (convertTimestamp time)
(E.UserProcess (fromIntegral receiverMachine) (fromIntegral receiverProcess))
(fromIntegral senderThread)
(fromIntegral receiverInport)
(readTag mesTag)),
nms, nps)
ReceiveMessage{
mesTag,
receiverProcess,
receiverInport,
senderMachine,
senderProcess,
senderThread,
messageSize} -> (True, createNewMsgEvents
(E.UserProcess (fromIntegral senderMachine) (fromIntegral senderProcess))
(E.ORM (convertTimestamp time)
(E.UserProcess mId (fromIntegral receiverProcess))
(fromIntegral senderThread)
(fromIntegral receiverInport)
(readTag mesTag)
(fromIntegral messageSize)),
nms, nps)
{- SendReceiveLocalMessage{
mesTag,
senderProcess,
senderThread,
receiverProcess,
receiverInport} -> trace "insertEvent SRLM (unhandled)" (False, undefined, undefined, undefined) --will be something like (True, newOPenMEssageList, nms, nps)
--unfinished ... #todo
-}
_ -> (False, undefined, undefined, undefined)
where createNewMsgEvents i event = insertMessage i event ocMsgs
proc = case spec of SendMessage _ p _ _ _ _ -> fromIntegral p; ReceiveMessage _ p _ _ _ _ _ -> fromIntegral p
(nms, nps) = increaseMsgCount (E.UserProcess mId proc) ms ps spec
(isTrdEvent,newTrdEvents) = case spec of
RunThread {
thread } -> (True, newEvents (E.RunThread (convertTimestamp time)))
CreateThread {
thread } -> (True, newEvents (E.NewThread (convertTimestamp time) (0))) -- #####hack outport? always zero in toSDDF
StopThread {
thread,
status} -> res where
res | status == ThreadFinished = (True, newEvents (E.KillThread (convertTimestamp time)))
| status `elem` blockList = (True, newEvents (E.BlockThread (convertTimestamp time) (0) status)) -- ####hack inport? reason?
| otherwise = (True, newEvents (E.SuspendThread (convertTimestamp time)))
blockList = [ThreadBlocked,BlockedOnMVar,BlockedOnBlackHole,BlockedOnDelay,BlockedOnSTM,
BlockedOnDoProc,BlockedOnMsgThrowTo,ThreadMigrating,BlockedOnMsgGlobalise,BlockedOnBlackHoleOwnedBy 0]
WakeupThread {
thread,
otherCap} -> (True, newEvents (E.DeblockThread (convertTimestamp time)))
_ -> (False, undefined)
where proc = lookupProcess lutable mId thre -- hack
thre = fromIntegral (thread spec) -- hack
newEvents evt = let pid = if proc == -1 then E.System mId
else E.UserProcess mId proc
in insertThreadEvent (pid, thre) evt ts
(isPrcEvent,newPrcEvents,newOcMsgs,newNP) = case spec of
CreateProcess {
process} -> (True, newEvents (E.NewProcess (convertTimestamp time)), ocMsgs, nP + 1)
KillProcess {
process} -> (True, newEvents (E.KillProcess (convertTimestamp time) (0,0,0)), delFromProcList (E.UserProcess mId (fromIntegral process)) ocMsgs, nP)
_ -> (False, undefined, undefined, undefined)
where newEvents evt = insPEvent (E.UserProcess mId (fromIntegral (process spec))) evt ps
(isMchEvent,newMchEvents,newMchTimes,maxLD',gcEvents) = case spec of
CreateMachine{} -> (True, newEvents (E.StartMachine (convertTimestamp time)), (mId, convertTimestamp time):mts, maxLD, Nothing)
Startup{n_caps} -> (True, ms, mts, maxLD, Nothing)
KillMachine _ -> (True, newEvents (E.EndMachine (convertTimestamp time)), mts, maxLD, Nothing)
-- JB, WAS: 849 ->
--233 -> (True, newEvents (GCMachine getGCTime v2 v3 v4 v5), mts, max maxLD v5, -- hack todo GC
-- Just (GCProcess getGCTime v2 v3 v4 v5, GCThread getGCTime v2 v3 v4 v5))
_ -> (False, undefined, undefined, undefined, undefined)
where newEvents evt = insMEvent mId evt ms
-- compute new min/max times:
newMax = max (convertTimestamp time) maxTime
(min', max') = if (convertTimestamp time) > maxTime
then (minTime, (convertTimestamp time))
else (newMin, maxTime)
newMin :: E.Seconds
newMin = min (convertTimestamp time) minTime
insMEventList :: E.MachineID -> [E.MachineEvent] -> [E.Machine] -> [E.Machine]
insMEventList i (m:ms) lst = insMEvent i m (insMEventList i ms lst)
insMEventList _ _ lst = lst
insMEvent :: E.MachineID -> E.MachineEvent -> [E.Machine] -> [E.Machine]
--insMEvent i1 evt lst@(e@(i2,allP,blkP,stat@(p,s,r),evts):es) -- insert in existing list of machines
insMEvent i1 evt (m:ms) -- insert in existing list of machines
| E.getIdM m > i1 = let ms' = (insMEvent i1 evt ms) in seq ms' (m : ms') -- go on
| E.getIdM m == i1 = m' `seq` (m' : ms) -- existing machine
| otherwise = (E.newMachine i1) {E.eventlistM = [evt]} : m : ms -- new machine => no processes
where
evts = E.eventlistM m
m' = case evt of
E.MSuspendProcess sec -> let newRunP = E.runningProcesses m - 1
newEvts = (E.SuspendedMachine sec):evts
in m {E.eventlistM = newEvts,
E.runningProcesses = newRunP}
E.MRunProcess sec -> let newRunP = E.runningProcesses m + 1
in m {E.eventlistM = (E.RunningMachine sec):evts,
E.runningProcesses = newRunP}
E.MBlockProcess sec -> let newRunP = E.runningProcesses m - 1
newBlkP = E.blockedProcesses m + 1
newEvts = if newRunP == 0
then (E.BlockedMachine sec):evts
else evts
in m {E.runningProcesses = newRunP,
E.blockedProcesses = newBlkP,
E.eventlistM = newEvts}
E.MDeblockProcess sec -> let newBlkP = E.blockedProcesses m - 1
newEvts = if (E.blockedProcesses m == E.aliveProcesses m)
then (E.SuspendedMachine sec):evts
else evts
in m {E.blockedProcesses = newBlkP,
E.eventlistM = newEvts}
E.GCMachine _ _ _ _ _ -> m
E.MNewProcess sec -> let newAllP = E.aliveProcesses m + 1
newTotP = E.totalProcesses m + 1
newEvts = if (E.blockedProcesses m == E.aliveProcesses m) -- test if evt may be skipped:
then (E.SuspendedMachine sec):evts
else evts
in m {E.aliveProcesses = newAllP,
E.totalProcesses = newTotP,
E.eventlistM = newEvts}
E.MKillRProcess sec -> let newRunP = E.runningProcesses m - 1
newAllP = E.aliveProcesses m - 1
newEvts | newRunP > 0 = evts -- machine keeps on running
| newAllP == 0 = (E.IdleMachine sec):evts
| E.blockedProcesses m == newAllP = (E.BlockedMachine sec):evts
| otherwise = (E.SuspendedMachine sec):evts
in m {E.aliveProcesses = newAllP,
E.runningProcesses = newRunP,
E.eventlistM = newEvts}
E.MKillSProcess sec -> let newAllP = E.aliveProcesses m - 1
newEvts | newAllP == 0 = (E.IdleMachine sec):evts
| newAllP == E.blockedProcesses m = (E.BlockedMachine sec):evts
| otherwise = evts
in m {E.aliveProcesses = newAllP,
E.eventlistM = newEvts}
E.MKillBProcess sec -> let newAllP = E.aliveProcesses m - 1
newBlkP = E.blockedProcesses m - 1
newEvts = if newAllP > 0
then evts
else (E.IdleMachine sec):evts
in m {E.aliveProcesses = newAllP,
E.blockedProcesses = newBlkP,
E.eventlistM = newEvts}
E.MIdleProcess sec -> let newRunP = E.runningProcesses m - 1
newEvts = if newRunP == 0
then (E.IdleMachine sec) : evts
else evts
in m {E.runningProcesses = newRunP}
E.EndMachine _ -> m {E.runningProcesses = 0,
E.aliveProcesses = 0,
E.blockedProcesses = 0,
E.eventlistM = evt:evts}
_ -> error ("insMEvent: unknown event: " ++ show evt)
insMEvent i1 evt [] = [(E.newMachine i1) {E.eventlistM = [evt]}] -- brand new machine => no processes
insPEventList :: [(E.ProcessID,E.ProcessEvent)] -> [E.Process] -> ([E.Process],[E.MachineEvent])
insPEventList ((i,p):ps) lst = let (ps',mes) = insPEventList ps lst
(lst',me) = insPEvent i p ps'
in case me of
Nothing -> (lst',mes)
Just me -> (lst',me:mes)
insPEventList _ lst = (lst,[])
insPeByMID :: E.MachineID -> E.ProcessEvent -> [E.Process] -> [E.Process]
insPeByMID i evt lst@(e:es)
| i == E.getMIdFromP e = insertHere lst -- first process on machine i found, insert event
| otherwise = let es' = insPeByMID i evt es in seq es' (e:es') -- search on
where insertHere lst@(e:es)
| i == E.getMIdFromP e = let es' = insertHere es
evts = E.eventlistP e
in case take 1 evts of
[E.KillProcess _ _] -> seq es' ( e : es') -- process not alive
[E.BlockedProcess _] -> seq es' (e {E.eventlistP = E.BlockedProcess (convertTimestamp time):evt:evts}:es')
[E.SuspendedProcess _] -> seq es' (e {E.eventlistP = E.SuspendedProcess (convertTimestamp time):evt:evts}:es')
_ -> seq es' (e {E.eventlistP = evt:evts} : es') -- not possible?
| otherwise = lst -- all processes on machine i worked up
insertHere [] = []
insPEvent :: E.ProcessID -> E.ProcessEvent -> [E.Process] -> ([E.Process],Maybe E.MachineEvent)
insPEvent i1 evt lst@(e:es) --e@(i2,allT,blkT,stat@(t,s,r),evts)
| i1 < E.getIdP e = let (es',mEvt) = insPEvent i1 evt es in seq es' (seq mEvt (e : es', mEvt))
| i1 == E.getIdP e = (insertHere : es, vMachineEvent) {- case evt of -- a new entry for every new process
NewProcess sec -> ((i1,0,0,(0,0,0),[evt]):lst, Just (MNewProcess time))
_ -> (insertHere : es, vMachineEvent) -}
-- | otherwise = ((i1,0,0,(0,0,0),[evt]):lst,Just (E.MNewProcess (convertTimestamp time))) -- new Process => evt == NewProcess
| otherwise = insPEvent i1 evt (E.newProcess i1 : lst) -- evt for new process => create it and handle event normally
where (insertHere,vMachineEvent) = let allT = E.aliveThreads e
runT = E.runningThreads e
blkT = E.blockedThreads e
evts = E.eventlistP e
in case evt of
E.PNewThread sec -> let !newAllT = allT + 1
!newTotT = E.totalThreads e + 1
!newEvts = if blkT == allT --process was blocked or is newborn
then (E.SuspendedProcess sec) : evts
else evts
!newMEvt = if evts == [] -- process is newborn
then Just (E.MNewProcess sec) -- inform machine here... CreateProcess event will follow directly only if this is not system thread
else if blkT == allT -- process was blocked
then Just (E.MDeblockProcess sec)
else Nothing
in (e {E.aliveThreads = newAllT,
E.totalThreads = newTotT,
E.eventlistP = newEvts},
newMEvt)
E.PKillRThread sec -> let !newAllT = allT - 1
!newRunT = runT - 1
!(newEvts,newMEvt) = if newAllT > 0
then if E.blockedThreads e < newAllT -- process was running, now suspended or blocked?
then ((E.SuspendedProcess sec):evts, Just (E.MSuspendProcess sec))
else ((E.BlockedProcess sec):evts, Just (E.MBlockProcess sec))
else case evts of
(E.KillProcess _ _:_) -> (evts, Nothing)
_ -> case E.getIdP e of
E.UserProcess _ _ -> ((E.IdleProcess sec):evts, Just (E.MIdleProcess sec))
E.System _ -> (E.KillProcess sec (E.totalThreads e, 0, 0):evts, Just (E.MKillRProcess sec))
in (e {E.aliveThreads = newAllT,
E.runningThreads = newRunT,
E.eventlistP = newEvts},
newMEvt)
E.PKillSThread sec -> let !newAllT = allT - 1
!(newEvts,newMEvt) = if newAllT > 0
then if blkT < newAllT -- process was suspended, now still suspended or blocked?
then (evts, Nothing)
else ((E.BlockedProcess sec):evts, Just (E.MBlockProcess sec))
else case evts of
(E.KillProcess _ _:_) -> (evts, Nothing)
_ -> case E.getIdP e of
E.UserProcess _ _ -> ((E.IdleProcess sec):evts, Just (E.MIdleProcess sec))
E.System _ -> (E.KillProcess sec (E.totalThreads e, 0, 0):evts, Just (E.MKillSProcess sec))
in (e {E.aliveThreads = newAllT,
E.eventlistP = newEvts},
newMEvt)
E.PKillBThread sec -> let !newAllT = allT - 1
!newBlkT = blkT - 1
!(newEvts,newMEvt) = if newAllT > 0
then (evts, Nothing)
else case evts of
(E.KillProcess _ _:_) -> (evts, Nothing)
_ -> case E.getIdP e of
E.UserProcess _ _ -> ((E.IdleProcess sec):evts, Just (E.MIdleProcess sec))
E.System _ -> (E.KillProcess sec (E.totalThreads e, 0, 0):evts, Just (E.MKillBProcess sec))
in (e {E.aliveThreads = newAllT,
E.blockedThreads = newBlkT,
E.eventlistP = newEvts},
newMEvt)
E.PRunThread sec -> let !newRunT = runT + 1
!newEvts = (E.RunningProcess sec):evts
in (e {E.runningThreads = newRunT,
E.eventlistP = newEvts},
Just (E.MRunProcess sec))
E.PSuspendThread sec -> let !newRunT = runT - 1
!newEvts = (E.SuspendedProcess sec):evts
in (e {E.runningThreads = newRunT,
E.eventlistP = newEvts},
Just (E.MSuspendProcess sec))
E.PBlockThread sec -> let !newBlkT = blkT + 1
!newRunT = runT - 1
!(newEvts,newMEvt) = if newBlkT < allT
then if newRunT > 0
then (evts, Nothing)
else ((E.SuspendedProcess sec):evts, Just (E.MSuspendProcess sec))
else ((E.BlockedProcess sec):evts,Just (E.MBlockProcess sec))
in (e {E.blockedThreads = newBlkT,
E.runningThreads = newRunT,
E.eventlistP = newEvts},
newMEvt)
E.PDeblockThread sec -> let !newBlkT = blkT - 1
!(newEvts,newMEvt) = if blkT == allT -- was process blocked?
then ((E.SuspendedProcess sec):evts,Just (E.MDeblockProcess sec))
else (evts,Nothing)
in (e {E.blockedThreads = newBlkT,
E.eventlistP = newEvts},
newMEvt)
E.NewProcess sec -> (e {E.eventlistP = evt:evts}, Nothing) -- note that MNewProcessEvent was already generated for preceding PNewThread event
E.LabelProcess sec _ -> (e {E.eventlistP = evt:evts}, Nothing)
-- KillProcess holds the statistic information for the ending process
E.KillProcess sec _ -> let stat = (E.totalThreads e, E.sentMessagesP e, E.receivedMessagesP e)
evt' = E.KillProcess sec stat
e' = (E.newProcess (E.getIdP e)) {E.eventlistP = evt':evts}
in case evts of
(E.RunningProcess _:_) -> (e',Just (E.MKillRProcess sec))
(E.BlockedProcess _:_) -> (e',Just (E.MKillBProcess sec))
_ -> (e',Just (E.MKillSProcess sec))
_ -> error ("unknown event: " ++ show evt)
-- insPEvent i1 evt [] = ([(i1,0,0,(0,0,0),[evt])],Just (E.MNewProcess (convertTimestamp time))) -- new Process => evt == NewProcess
insPEvent i1 evt [] = insPEvent i1 evt [E.newProcess i1] -- evt for new process => create it and handle event normally
insTeByMID :: E.MachineID -> E.ThreadEvent -> [E.OpenThread] -> [E.OpenThread]
insTeByMID i evt lst@(e@(m,(lti,lte),ts):es)
| i == m = (m,((E.UserProcess m (-1),-1),evt),insertHere ts):es -- machine i found, insert event into threads; the ((E.UserProcess m (-1),-1)
-- inserts 'evt' the next time insTEvent is run.
| otherwise = let es' = insTeByMID i evt es in seq es' (e:es') -- search on
where insertHere (l@(i,evts):ls) = let ls' = insertHere ls
in case take 1 evts of
-- skip already killed threads:
[E.KillThread _] -> seq ls' (l:ls') -- Thread already dead
-- The following entry describes the last suspended thread. The SuspendThread-event hasn't yet
-- been inserted, it resides in 'lte':
[E.RunThread _] -> seq ls' ((i,E.setEventTime lte (convertTimestamp time):evt:lte:evts):ls')
-- other threads are blocked or suspended:
[lastEvent] -> seq ls' ((i,E.setEventTime lastEvent (convertTimestamp time):evt:evts):ls')
insertHere [] = []
insertThreadEvent :: E.ThreadID -> E.ThreadEvent -> [E.OpenThread] -> ([E.OpenThread], [(E.ProcessID,E.ProcessEvent)])
insertThreadEvent i@(pid,_) evt tl@(t@(im,(lti,lte),ts):tls)
| E.pId2mId pid < im = let (tls', pEvt') = insertThreadEvent i evt tls -- look for corresponding machine
in seq tls' (seq pEvt' (t:tls',pEvt')) -- recurse
| E.pId2mId pid == im = case lte of
E.SuspendThread _ -> case evt of
E.RunThread _ -> if i == lti
then ((im,(i,E.DummyThread),ts):tls,[]) -- suppress flattering
else bothEvents
_ -> bothEvents
E.DummyThread -> case evt of
E.SuspendThread _ -> ((im,(i,evt),ts):tls, []) -- deter SuspendThread-Event
_ -> let (ts',pEvt') = insTEvent' i evt ts in ((im,(i,E.DummyThread),ts'):tls,[(t2pID i,pEvt')])
otherwise -> trace "should this really happen?" $ insertThreadEvent i evt tls
| otherwise = ((E.pId2mId pid,(i,E.DummyThread),[(i,[evt])]):tl,[(t2pID i,vProcessEvent evt [])]) -- new machine
where bothEvents = let (ts2,pEvt2) = insTEvent' lti lte ts
(ts3,pEvt3) = insTEvent' i evt ts2
in ((im,(i,E.DummyThread),ts3):tls,[(t2pID i,pEvt3),(t2pID lti,pEvt2)])
insertThreadEvent i@(pid,_) evt [] = ([(E.pId2mId pid,(i,E.DummyThread),[(i,[evt])])],[(t2pID i,vProcessEvent evt [])])
insTEvent' :: E.ThreadID -> E.ThreadEvent -> [E.Thread] -> ([E.Thread],E.ProcessEvent)
insTEvent' i@(pid,t) ne lst@(e@(ib@(ipid,it),evts):es)
| i' < ib' = let (es', mEvt') = (insTEvent' i ne es) in seq es' (seq mEvt' (e:es',mEvt'))
| i' == ib' = ((i,(ne:evts)):es, vProcessEvent ne evts)
| otherwise = ((i,[ne]):lst,vProcessEvent ne evts)
where
i' = (E.pId2mId pid,t)
ib' = (E.pId2mId ipid,it)
insTEvent' i evt _ = ([(i,[evt])],vProcessEvent evt [])
vProcessEvent :: E.ThreadEvent -> [E.ThreadEvent] -> E.ProcessEvent
vProcessEvent evt evts = case evt of
E.KillThread sec -> case evts of
(E.BlockThread _ _ _:_) -> E.PKillBThread sec
(E.RunThread _ :_) -> E.PKillRThread sec
_ -> E.PKillSThread sec
E.RunThread sec -> E.PRunThread sec
E.SuspendThread sec -> E.PSuspendThread sec
E.BlockThread sec _ _ -> E.PBlockThread sec
E.DeblockThread sec -> E.PDeblockThread sec
E.NewThread sec _ -> E.PNewThread sec
increaseMsgCount :: E.ProcessID -> [E.Machine] -> [E.Process] ->
EventInfo -> ([E.Machine], [E.Process])
increaseMsgCount pID ml pl spec = (incM ml, incP pl)
where incM :: [E.Machine] -> [E.Machine]
--incM (m@(i, aP, bP, (p,s,r), es):ms)
incM (m:ms)
| (E.getIdM m) > (E.pId2mId pID) = m : incM ms -- go on
| (E.getIdM m) == (E.pId2mId pID) = case spec of
SendMessage _ _ _ _ _ _ -> m {E.sentMessagesM = E.sentMessagesM m + 1} : ms
ReceiveMessage _ _ _ _ _ _ _ -> m {E.receivedMessagesM = E.receivedMessagesM m + 1} : ms
SendReceiveLocalMessage {} -> m {E.sentMessagesM = E.sentMessagesM m + 1,
E.receivedMessagesM = E.receivedMessagesM m + 1} : ms
otherwise -> m:ms -- should not occur
| otherwise = m:ms -- not found? nevermind...
incM _ = []
incP :: [E.Process] -> [E.Process]
incP (p:ps)
| E.getIdP p > pID = p : incP ps
| E.getIdP p == pID = case spec of
SendMessage _ _ _ _ _ _ -> p {E.sentMessagesP = E.sentMessagesP p + 1} : ps
ReceiveMessage _ _ _ _ _ _ _ -> p {E.receivedMessagesP = E.receivedMessagesP p + 1} : ps
SendReceiveLocalMessage {} -> p {E.sentMessagesP = E.sentMessagesP p + 1,
E.receivedMessagesP = E.receivedMessagesP p + 1} : ps
otherwise -> p:ps -- should not occur
| otherwise = p:ps -- not found? nevermind...
incP _ = []
newProc :: E.ProcessID -> E.ProcessID -> E.ProcessList -> E.ProcessTree -> (E.ProcessList, E.ProcessTree)
newProc dad son pls pt = (addProcPath son (dadPath) pls, addChildPrc son dadPath pt)
where dadPath = []--TODO getPath dad pls ++ [dad]
getPath :: E.ProcessID -> E.ProcessList -> [E.ProcessID]
getPath pId (p@(i,path):ps)
| i < pId = getPath pId ps
| i == pId = path
| otherwise = error ("not impl: getPath, otherwise " ++ (show son) ++ " " ++ show (pId,pls))
getPath _ _ = []
addProcPath :: E.ProcessID -> [E.ProcessID] -> E.ProcessList -> E.ProcessList
addProcPath pId path pls@(p@(i,lst):ps)
| i < pId = let ps' = addProcPath pId path ps in seq ps' (p:ps')
| i == pId = (pId,path) : ps
| otherwise = (pId,path) : pls
addProcPath pId path _ = [(pId, path)]
addChildPrc :: E.ProcessID -> [E.ProcessID] -> E.ProcessTree -> E.ProcessTree
addChildPrc pId [p] (Node i pts) = Node i ((Node pId []):pts)
addChildPrc pId path@(p:ps) pt@(Node i pts)
| p == i = let pts' = stepDown ps pts in seq pts' (Node i pts')
| otherwise = error ("addChildPrc: wrong ProcessTree found (" ++ show p ++ "!=" ++ show i ++ ")")
where stepDown :: [E.ProcessID] -> [E.ProcessTree] -> [E.ProcessTree]
stepDown path@(p:ps) pts@(t@(Node i _):ts)
| i == p = let t' = addChildPrc pId path t in seq t' (t':ts)
| otherwise = let ts' = stepDown path ts in seq ts' (t:ts')
stepDown path [] = seq (putStrLn ("stepDown: path " ++ show path ++ " not found in Process Tree:" ++ show pt)) []
addChildPrc pId [] pt = pt
delFromProcList :: E.ProcessID -> E.OpenMessageList -> E.OpenMessageList
delFromProcList pId oml = oml
insertMessage :: E.ProcessID -> E.OpenMessageEvent -> E.OpenMessageList -> E.OpenMessageList
insertMessage sp newMessage ocMsgs@(openMsgList,closedMessages,partMsgs@(openPrcMsgs,prcTbl,prcTree),(headMessages,hSize,closedHeads))
| tag == E.Head = let ohl' = addHeadMsg (sp, out, rp, inp) headMessages
in seq ohl' (openMsgList, closedMessages, partMsgs,(ohl', hSize, closedHeads))
| tag == E.DataMes = let (ohl,chl,hs') = searchHeadMsg (sp,out,rp,inp) headMessages
in cml `seq` chl `seq` (oml,cml,partMsgs,(ohl,max hs' hSize,chl))
| tag == E.RFork = ocMsgs
| otherwise = let sp' = E.pId sp
rp' = E.pId rp
in if (min sp' rp') >= 0
then seq oml (seq cml (oml,cml,partMsgs,(headMessages,hSize,closedHeads)))
else ocMsgs
where (time, rp,out,inp,tag,size) = case newMessage of
E.ORM t p o i r s -> (t,p,o,i,r,s)
E.OSM t p o i r -> (t,p,o,i,r,0)
(oml,cml) = searchID sp newMessage openMsgList closedMessages
addHeadMsg :: E.ChannelID -> [E.OpenHeadMessage] -> [E.OpenHeadMessage]
addHeadMsg cId hml@(h@(idB,s,i,hsm,hrm):hs)
| cId > idB = let hs' = addHeadMsg cId hs in seq hs' (h:hs') -- not found yet: search on
| cId == idB = case newMessage of
E.ORM _ _ _ _ _ _ -> case hrm of -- entry found, increase quantity and size
[firstMsg] -> (idB,(s+size),(i+1),hsm,time:hrm):hs -- second received Message
(lastMsg:fm) -> (idB,(s+size),(i+1),hsm, time:fm):hs -- replace last received Message
[] -> (idB,(s+size),(i+1),hsm, [time] ):hs -- first received Message
E.OSM _ _ _ _ _ -> case hsm of -- entry found, don't touch values
[firstMsg] -> (idB,s,i,time:hsm,hrm):hs
(lastMsg:fm) -> (idB,s,i,time:fm ,hrm):hs
[] -> (idB,s,i, [time] ,hrm):hs
| otherwise = case newMessage of
E.ORM _ _ _ _ _ _ -> (cId,size,1,[],[time]):hml -- insert new entry before h
E.OSM _ _ _ _ _ -> (cId,size,0,[time],[]):hml
addHeadMsg cId [] = case newMessage of
E.ORM _ _ _ _ _ _ -> [(cId,size,1,[],[time])]
E.OSM _ _ _ _ _ -> [(cId,size,0,[time],[])]
searchHeadMsg :: E.ChannelID -> [E.OpenHeadMessage] -> ([E.OpenHeadMessage],[E.HeadMessage],Double)
searchHeadMsg cId hml@(h@(idB,s,i,hsm,hrm):hs)
| cId > idB = let (hs',ch',ms') = searchHeadMsg cId hs
in seq hs' (seq ch' (h:hs',ch',ms'))
| cId == idB = let sInt = s + size
sDouble = fromIntegral sInt
in case newMessage of
E.ORM _ _ _ _ _ _ ->
if length hsm == 3
then (hs, (cId,(ts1,tr1,ts2,time),sDouble,(i+1)):closedHeads, sDouble)
else ((idB,sInt,(i+1),hsm,time:hrm):hs,closedHeads,sDouble)
E.OSM _ _ _ _ _ ->
if length hrm == 3
then (hs, (cId,(ts1,tr1,time,tr2),sDouble,i):closedHeads,sDouble)
else ((idB,s,i,time:hsm,hrm):hs,closedHeads,sDouble)
| otherwise = (hml,closedHeads,0)
where ts2 = head hsm
ts1 = last hsm
tr2 = head hrm
tr1 = last hrm
searchHeadMsg cId [] = ([], closedHeads,0)
-- insertClosedMessage :: Message -> OpenMessage -> OpenMessageList
-- insertClosedMessage m@(SendReceiveLocalMessage {}) (openMsgList,closedMessages,partMsgs,(headMessages,hSize,closedHeads)) = todo
closeOpenLists :: E.OpenEvents -> E.Events
closeOpenLists (events@(mEvents,pEvents,tEvents),mTimes,(_,closedMessages,(_,_,Node _ pTrees),(openHeadMessages,hSize,headMessages)),(minTime,maxTime,numP,maxLD)) =
mEvents `seq` minTime `seq` allHeadMessages `seq`
E.Events {
E.machinelist = mEvents,
E.processlist = pEvents,
E.threadlist = newThreadEvents,
E.starttimeByMachine = mTimes,
E.maxStartup = 0,
E.startupOffsets = undefined,
E.messagelist = (closedMessages,[],allHeadMessages,reversedProcTree, []),
E.min_t = minTime,
E.max_t = maxTime,
E.max_t_diff = 0,
E.maxMsgSize = hSize,
E.maxLD = fromIntegral maxLD,
E.noOfMachines = length mEvents,
E.noOfProcesses = numP,
E.noOfThreads = length newThreadEvents
}
-- ((mEvents,pEvents,newThreadEvents),mTimes,undefined,(closedMessages,[],allHeadMessages,reversedProcTree, []),
-- (minTime,maxTime,0,hSize,(fromIntegral maxLD)),(length mEvents,numP,length newThreadEvents))
where allHeadMessages = handleOpenHeadMsgs openHeadMessages headMessages
newThreadEvents = concat (map (\(_,_,ts) -> ts) tEvents)
reversedProcTree :: E.ProcessTree
reversedProcTree = if null pTrees
then Node (E.UserProcess 0 0) []
else reverseSubForests (head pTrees)
reverseSubForests :: E.ProcessTree -> E.ProcessTree
reverseSubForests (Node i f) = Node i (map reverseSubForests (reverse f))
handleOpenHeadMsgs :: [E.OpenHeadMessage] -> [E.HeadMessage] -> [E.HeadMessage]
handleOpenHeadMsgs [] hm = hm
handleOpenHeadMsgs ((ch,s,i,sml,rml):os) hm = case zip sml rml of
((lst',lrt'):(fst',frt'):_) -> handleOpenHeadMsgs os ((newHeadMsg fst' frt' lst' lrt'):hm)
_ -> handleOpenHeadMsgs os hm
where newHeadMsg :: E.Seconds -> E.Seconds -> E.Seconds -> E.Seconds -> E.HeadMessage
newHeadMsg tS1 tR1 tS2 tR2 = (ch,(tS1,tR1,tS2,tR2),fromIntegral s,i)
t2pID :: E.ThreadID -> E.ProcessID
t2pID tid = fst tid