bin/appliskel/tests/concurrentSession/TestMinimalExample.py

   1 #!/usr/bin/env python3
   2 # Copyright (C) 2013-2023  CEA, EDF, OPEN CASCADE
   3 #
   4 # This library is free software; you can redistribute it and/or
   5 # modify it under the terms of the GNU Lesser General Public
   6 # License as published by the Free Software Foundation; either
   7 # version 2.1 of the License, or (at your option) any later version.
   8 #
   9 # This library is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12 # Lesser General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU Lesser General Public
  15 # License along with this library; if not, write to the Free Software
  16 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  17 #
  18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  19 #
  20
  21 import os
  22 import sys
  23 import multiprocessing
  24 import unittest
  25 import logging
  26
  27
  28 def port_reservation(obtained_ports, preferred=None, name = None, messages=None, expected=None):
  29   from PortManager import getPort
  30   if preferred:
  31     port = getPort(preferred)
  32   else:
  33     port = getPort()
  34   print("obtained port = %s"%port)
  35
  36   obtained_ports.put(port)
  37
  38   if expected and messages:
  39     _name = name + " : " if name else ""
  40     if port != expected:
  41       messages.put(_name + "used port= %s, expected port = %s"%(port, expected))
  42     else:
  43       messages.put(_name + "OK")
  44 #
  45
  46
  47 class TestMinimalExample(unittest.TestCase):
  48
  49   def testSequential(self):
  50     from PortManager import releasePort, getBusyPorts
  51     print("\nBEGIN testSequential")
  52     print("Busy ports", getBusyPorts())
  53     obtained_ports = multiprocessing.Queue()
  54     messages = multiprocessing.Queue()
  55
  56     processes = [
  57       multiprocessing.Process(target=port_reservation, args=(obtained_ports,))
  58       for i in range(3)
  59       ]
  60
  61     for p in processes:
  62       p.start()
  63
  64     for p in processes:
  65       p.join()
  66
  67     print("Busy ports", getBusyPorts())
  68     # Try to get specific port number
  69     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2872, "testSequential 2872",
  70                                                                messages, 2872,))
  71     p.start()
  72     p.join()
  73
  74     # Try to get specific port number
  75     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))
  76     p.start()
  77     p.join()
  78
  79     # Try to get specific port number
  80     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2899, "testSequential 2899:1",
  81                                                                messages, 2899,))
  82     p.start()
  83     p.join()
  84
  85     # Release port
  86     print ("release port 2899")
  87     p = multiprocessing.Process(target=releasePort, args=(2899,))
  88     p.start()
  89     p.join()
  90
  91     # Try to get specific port number
  92     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2899, "testSequential 2899:2",
  93                                                                messages, 2899,))
  94     p.start()
  95     p.join()
  96
  97     # Check results
  98     while not messages.empty():
  99       message = messages.get()
 100       if "OK" not in message:
 101         self.fail(message)
 102
 103     # Release ports
 104     print("Busy ports", getBusyPorts())
 105     while not obtained_ports.empty():
 106       port = obtained_ports.get()
 107       print("release port", port)
 108       p = multiprocessing.Process(target=releasePort, args=(port,))
 109       p.start()
 110       p.join()
 111
 112     print("END testSequential")
 113   #
 114
 115   def testConcurrent(self):
 116     from PortManager import releasePort, getBusyPorts
 117     print("\nBEGIN testConcurrent")
 118     print("Busy ports", getBusyPorts())
 119     obtained_ports = multiprocessing.Queue()
 120     messages = multiprocessing.Queue()
 121     processes = [
 122       multiprocessing.Process(target=port_reservation, args=(obtained_ports,))
 123
 124       for i in range(3)
 125       ]
 126
 127     # Try to get specific port number
 128     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2872, "testSequential 2872",
 129                                                                messages, 2872,))
 130     processes.append(p)
 131
 132     # Try to get specific port number
 133     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))
 134     processes.append(p)
 135
 136     # Try to get specific port number
 137     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))
 138     processes.append(p)
 139
 140     for p in processes:
 141       p.start()
 142
 143     for p in processes:
 144       p.join()
 145
 146     # Check results
 147     while not messages.empty():
 148       message = messages.get()
 149       if "OK" not in message:
 150         self.fail(message)
 151
 152     # Release ports
 153     print("Busy ports", getBusyPorts())
 154     while not obtained_ports.empty():
 155       port = obtained_ports.get()
 156       print("release port", port)
 157       p = multiprocessing.Process(target=releasePort, args=(port,))
 158       p.start()
 159       p.join()
 160
 161     print("END testConcurrent")
 162   #
 163 #
 164
 165 if __name__ == "__main__":
 166   omniorb_user_path = os.getenv("OMNIORB_USER_PATH")
 167   if not omniorb_user_path:
 168     msg = "\n"
 169     msg += "Error: please set OMNIORB_USER_PATH variable.\n"
 170     msg += "       Usually this points to your application USERS directory.\n"
 171     logging.error(msg)
 172     sys.exit(1)
 173
 174   try:
 175     import PortManager
 176   except ImportError:
 177     msg = "\n"
 178     msg += "Error: can't import PortManager; please check PYTHONPATH variable.\n"
 179     msg += "       You need to add <KERNEL_INSTALL_PATH>/bin/salome path.\n"
 180     logging.error(msg)
 181     sys.exit(1)
 182
 183   unittest.main()
 184 #