1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
------------------------------------------------------------------------------
--- Implementation of a Curry Port Name Server based on raw sockets.
--- It is used to implement the library Ports for distributed programming
--- with ports.
---
--- @author Michael Hanus
--- @version February 2017
--- @category web
------------------------------------------------------------------------------

module CPNS(registerPort,getPortInfo,unregisterPort,
            cpnsStart,cpnsStop,cpnsShow,cpnsAlive,main) where

import Char
import Distribution(installDir)
import FilePath((</>))
import IO
import List(delete,intersperse)
import Profile
import ReadShowTerm
import Socket
import System
import Time

-- If we connect to a port with symbolic name pn, we first connect
-- to the CPNS of the host named by pn to get the physical socket
-- number of this port. In order to connect to CPNS from any
-- machine in the world, the CPNS demon always listens at the following
-- port:
-- (Note that this must be identical for all machines running
-- Distributed Curry! If this port is occupied by another process
-- on a host, you cannot run Distributed Curry on it.)

-- The standard port number of CPNS demon.
cpnsSocket :: Int
cpnsSocket = 8767


-- The time out before considering the server as unreachable:
cpnsTimeOut :: Int
cpnsTimeOut = 3000

--- Type of messages to be processed by the Curry Port Name Server.
---
--- @cons Register name pid sn pn ack
---       -  assign the values pid, sn, and pn to name
---          (pid is the process number of the registered process
---           (should be 0 if it is unknown); the server returns True
---           if registration had no problems, otherwise False)
--- @cons GetRegister name - request for a registered port name;
---         the server returns the values (sn,pn) that are assigned to the
---         port name
--- @cons Unregister name - request to remove a registered port name
--- @cons ShowRegistry - show the current port registrations
--- @cons Ping         - ping the CPNS demon for liveness check
--- @cons Terminate    - terminate the CPNS demon
data CPNSMessage = Terminate
                 | Ping
                 | Register String Int Int Int
                 | GetRegister String
                 | Unregister String
                 | ShowRegistry

-- The lock file to coordinate the startup of the CPNS demon:
cpnsStartupLockfile :: String
cpnsStartupLockfile = "/tmp/CurryPNSD.lock"

--- Starts the "Curry Port Name Server" (CPNS) running on the local machine.
--- The CPNS is responsible to resolve symbolic names for ports
--- into physical socket numbers so that a port can be reached
--- under its symbolic name from any machine in the world.

cpnsStart :: IO ()
cpnsStart = catch startup
                  (\_ -> putStrLn "FAILURE occurred during startup!" >>
                         deleteStartupLockfile >>
                         return Nothing) >>=
             maybe done (cpnsServer [])
 where
   deleteStartupLockfile = do
     putStrLn $ "Removing startup lock file \""++cpnsStartupLockfile++"\"..."
     system $ "rm -f "++cpnsStartupLockfile
     done

   startup = do
     putStrLn $ "Starting Curry Port Name Server on port " ++
                show cpnsSocket ++ "..."
     socket <- listenOn cpnsSocket
     deleteStartupLockfile
     pid <- getPID
     putStrLn $ "Curry Port Name Server is ready (PID: "++show pid++")."
     return (Just socket)

--- The main loop of the CPNS demon
cpnsServer :: [(String,Int,Int,Int)] -> Socket -> IO ()
cpnsServer regs socket = do
  (chost,stream) <- socketAccept socket
  --putStrLn $ "Connection from "++chost
  serveRequest chost stream
 where
   doIfLocalHost rhost action = do
     hostname <- getHostname
     if rhost `elem` ["localhost","localhost.localdomain",hostname]
        || take 8 rhost == "127.0.0."
      then action
      else do putStrLn $ "Illegal request received from host: " ++ rhost
              cpnsServer regs socket

   serveRequest rhost h = do
     msg <- readMsgLine h
     either
      (\line -> do putStrLn $ "ERROR: Illegal message:\n" ++ line
                   cpnsServer regs socket )
      (\m -> case m of
        Terminate -> doIfLocalHost rhost $ do
          hClose h
          putStrLn "CPNS demon terminated."
        Ping -> do
          hPutStrLn h (showQTerm ())
          hClose h
          cpnsServer regs socket
        Register pname pid sn pn -> doIfLocalHost rhost $ do
          (ack, newregs) <- tryRegisterPortName regs pname pid sn pn
          hPutStrLn h (showQTerm ack)
          hClose h
          cpnsServer newregs socket
        GetRegister pname -> do
          --putStrLn $ "Request for port name: " ++ pname
          (newregs,pinfo) <- getRegisteredPortName regs pname
          hPutStrLn h (showQTerm pinfo)
          hClose h
          cpnsServer newregs socket
        Unregister pname -> doIfLocalHost rhost $ do
          newregs <- unregisterPortName regs pname
          hClose h
          cpnsServer newregs socket
        ShowRegistry -> doIfLocalHost rhost $ do
          putStrLn "Currently registered port names:"
          newregs <- showAndCleanRegs regs
          hFlush stdout
          hClose h
          cpnsServer newregs socket )
      msg

tryRegisterPortName :: [(String,Int,Int,Int)] -> String -> Int -> Int -> Int
                    -> IO (Bool, [(String, Int, Int, Int)])
tryRegisterPortName regs name pid sn pn = do
  let nameregs = filter (\(n,_,_,_)->name==n) regs
  ack <- if null nameregs
         then return True
         else let (_,pid',_,_) = head nameregs in
              if pid'>0 && pid'/=pid
              -- we allow registration from the same process
              then doesProcessExists pid' >>= \pex -> return (not pex)
              else return True
  ctime <- getLocalTime
  putStrLn $ "Register port \""++name++"\": pid "++show pid++
             " / socket "++show sn++
             " / number "++show pn ++ " at " ++ calendarTimeToString ctime
  let newregs = (name,pid,sn,pn) : filter (\ (n,_,_,_)->name/=n) regs
  printMemInfo newregs
  hFlush stdout
  return (ack, newregs)

-- Delete all registrations for a given port name:
unregisterPortName :: [(String,Int,Int,Int)] -> String
                   -> IO [(String,Int,Int,Int)]
unregisterPortName regs name = do
  ctime <- getLocalTime
  putStrLn $ "Unregister port \""++name++"\" at "++calendarTimeToString ctime
  let newregs = filter (\ (n,_,_,_)->name/=n) regs
  printMemInfo newregs
  hFlush stdout
  return newregs

-- Get the socket number for a registered port name
-- (or (0,0) if not registered).
-- In addition, a new registration list is returned where a registration
-- is deleted if the corresponding server process does not exist.
getRegisteredPortName :: [(String,Int,Int,Int)] -> String
                      -> IO ([(String,Int,Int,Int)],(Int,Int))
getRegisteredPortName regs pname =
  let nameregs = filter (\(n,_,_,_)->pname==n) regs in
  if null nameregs
  then return (regs,(0,0))
  else let (_,pid,sn,pn) = head nameregs in
       if pid>0
       then doesProcessExists pid >>= \pex ->
            if pex
            then return (regs,(sn,pn))
            else --putStrLn ("WARNING: Process "++show pid++" not running!") >>
                 return (delete (head nameregs) regs, (0,0))
       else return (regs,(sn,pn))

-- Show all registered ports and return a new registration list
-- where a registration is deleted if the corresponding server process
-- does not exist.
showAndCleanRegs :: [(String,Int,Int,Int)] -> IO [(String,Int,Int,Int)]
showAndCleanRegs regs = do
  newreglist <- mapIO checkAndShow regs
  return (concat newreglist)
 where
  checkAndShow reg@(n,pid,sn,pn) = do
    pidexist <- doesProcessExists pid
    if pidexist
     then do putStrLn $ n++": pid "++show pid++
                        " / socket "++show sn++" / number "++show pn
             return [reg]
     else return []

-- Print status information of current CPNS demon process:
printMemInfo :: [(String,Int,Int,Int)] -> IO ()
printMemInfo regs = do
  pinfos <- getProcessInfos
  putStrLn ("NumRegs: " ++ show (length regs) ++ ", " ++ showMemInfo pinfos)

-- test whether a process with a given pid is running:
doesProcessExists :: Int -> IO Bool
doesProcessExists pid = do
  status <- system("test -z \"`ps -p "++show pid++" | fgrep "++show pid++"`\"")
  return (status>0)

-- Read a line from a stream and check syntactical correctness of message:
readMsgLine :: Handle -> IO (Either String a)
readMsgLine handle = do
  line <- hGetLine handle
  case readsQTerm line of
     [(msg,rem)] -> return (if all isSpace rem then Right msg else Left line)
     _ -> return (Left line)

-- Perform an action if the CPNS demon at a given host is alive:
doIfAlive :: String -> IO a -> IO a
doIfAlive host action = do
  alive <- cpnsAlive cpnsTimeOut host
  if not alive
   then error $ "Curry port name server at host \""++host++
                "\" is not reachable (timeout after "++show cpnsTimeOut++
                " ms)!"
   else action

sendToLocalCPNS :: CPNSMessage -> IO ()
sendToLocalCPNS msg = doIfAlive "localhost" $ do
  h <- connectToSocket "localhost" cpnsSocket
  hPutStrLn h (showQTerm msg)
  hClose h

--- Shows all registered ports at the local CPNS demon (in its logfile).
cpnsShow :: IO ()
cpnsShow = sendToLocalCPNS ShowRegistry

--- Terminates the local CPNS demon
cpnsStop :: IO ()
cpnsStop = sendToLocalCPNS Terminate

--- Gets an answer from a Curry port name server on a host,
--- or reports an error.
cpnsTryGetAnswer :: String -> CPNSMessage -> IO _
cpnsTryGetAnswer host msg = catch tryGetAnswer connectError
 where
  tryGetAnswer = do
    h <- connectToSocket host cpnsSocket
    hPutStrLn h (showQTerm msg)
    hFlush h
    ready <- hWaitForInput h cpnsTimeOut
    if ready
     then do
       answer <- readMsgLine h
       hClose h
       either (\line -> error $ "cpnsTryGetAnswer: Illegal answer: " ++ line)
              return
              answer
     else failed

  connectError _ = error $ "Curry port name server at host \""++host++
                           "\" is not reachable!"

--- Registers a symbolic port at the local host.
registerPort :: String -> Int -> Int -> IO ()
registerPort pname sn pn = do
  startCPNSDIfNecessary
  pid <- getPID
  ack <- cpnsTryGetAnswer "localhost" (Register pname pid sn pn)
  if ack then done
         else putStrLn ("WARNING: Port name '"++pname++"' already registered!")

--- Gets the information about a symbolic port at some host.
getPortInfo :: String -> String -> IO (Int,Int)
getPortInfo pname host = cpnsTryGetAnswer host (GetRegister pname)

--- Unregisters a symbolic port at the local host.
unregisterPort :: String -> IO ()
unregisterPort pname = sendToLocalCPNS (Unregister pname)

--- Tests whether the CPNS demon at a host is alive.
cpnsAlive :: Int -> String -> IO Bool
cpnsAlive timeout host = catch tryPingCPNS (\_ -> return False)
 where
  tryPingCPNS = do
    h <- connectToSocket host cpnsSocket
    hPutStrLn h (showQTerm Ping)
    hFlush h
    answer <- hWaitForInput h timeout
    hClose h
    return answer

--- Starts the CPNS demon at localhost if it is not already running:
startCPNSDIfNecessary :: IO ()
startCPNSDIfNecessary = do
  system $ installDir </> "currytools" </> "cpns" </> "start"
  done

--- Main function for CPNS demon. Check arguments and execute command.
main :: IO ()
main = do
  args <- getArgs
  case args of
   ["start"] -> cpnsStart
   ["stop"]  -> cpnsStop
   ["show"]  -> cpnsShow
   _ -> putStrLn $ "ERROR: Illegal arguments: " ++
                   concat (intersperse " " args) ++ "\n" ++
                   "Allowed arguments: start|stop|show"

{-
Test with PAKCS:

:fork cpnsStart
registerPort "xxx" 42 2
getPortInfo "xxx" "localhost"
cpnsStop

-}