bin/appliskel/tests/concurrentSession/TestMinimalExample.py

   1 #!/usr/bin/env python3
   2 # Copyright (C) 2013-2020  CEA/DEN, EDF R&D, 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 def port_reservation(obtained_ports, preferred=None, test=None, expected=None):
  28   from PortManager import getPort
  29   if preferred:
  30     port = getPort(preferred)
  31   else:
  32     port = getPort()
  33   print("obtained port = %s"%port)
  34
  35   obtained_ports.put(port)
  36
  37   if expected:
  38     test.assertTrue(port == expected, "used = %s, expected = %s"%(port, expected))
  39 #
  40
  41 class TestMinimalExample(unittest.TestCase):
  42   def testSequential(self):
  43     from PortManager import releasePort, getBusyPorts
  44     print("\nBEGIN testSequential")
  45     print("Busy ports", getBusyPorts())
  46     obtained_ports = multiprocessing.Queue()
  47
  48     processes = [
  49       multiprocessing.Process(target=port_reservation, args=(obtained_ports,))
  50       for i in range(3)
  51       ]
  52
  53     for p in processes:
  54       p.start()
  55
  56     for p in processes:
  57       p.join()
  58
  59     print("Busy ports", getBusyPorts())
  60     # Try to get specific port number
  61     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2872, self, 2872,))
  62     p.start()
  63     p.join()
  64
  65     # Try to get specific port number
  66     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812, self,))
  67     p.start()
  68     p.join()
  69
  70     # Try to get specific port number
  71     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2899, self, 2899,))
  72     p.start()
  73     p.join()
  74
  75     # Release port
  76     print ("release port 2899")
  77     p = multiprocessing.Process(target=releasePort, args=(2899,))
  78     p.start()
  79     p.join()
  80
  81     # Try to get specific port number
  82     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2899, self, 2899,))
  83     p.start()
  84     p.join()
  85
  86     # Release ports
  87     print("Busy ports", getBusyPorts())
  88     while not obtained_ports.empty():
  89       port = obtained_ports.get()
  90       print("release port", port)
  91       p = multiprocessing.Process(target=releasePort, args=(port,))
  92       p.start()
  93       p.join()
  94
  95     print("END testSequential")
  96   #
  97
  98   def testConcurrent(self):
  99     from PortManager import releasePort, getBusyPorts
 100     print("\nBEGIN testConcurrent")
 101     print("Busy ports", getBusyPorts())
 102     obtained_ports = multiprocessing.Queue()
 103     processes = [
 104       multiprocessing.Process(target=port_reservation, args=(obtained_ports,))
 105
 106       for i in range(3)
 107       ]
 108
 109     # Try to get specific port number
 110     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2872, self, 2872,))
 111     processes.append(p)
 112
 113     # Try to get specific port number
 114     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))
 115     processes.append(p)
 116
 117     # Try to get specific port number
 118     p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))
 119     processes.append(p)
 120
 121     for p in processes:
 122       p.start()
 123
 124     for p in processes:
 125       p.join()
 126
 127     # Release ports
 128     print("Busy ports", getBusyPorts())
 129     while not obtained_ports.empty():
 130       port = obtained_ports.get()
 131       print("release port", port)
 132       p = multiprocessing.Process(target=releasePort, args=(port,))
 133       p.start()
 134       p.join()
 135
 136     print("END testConcurrent")
 137   #
 138 #
 139
 140 if __name__ == "__main__":
 141   omniorb_user_path = os.getenv("OMNIORB_USER_PATH")
 142   if not omniorb_user_path:
 143     msg = "\n"
 144     msg += "Error: please set OMNIORB_USER_PATH variable.\n"
 145     msg += "       Usually this points to your application USERS directory.\n"
 146     logging.error(msg)
 147     sys.exit(1)
 148
 149   try:
 150     import PortManager
 151   except ImportError:
 152     msg = "\n"
 153     msg += "Error: can't import PortManager; please check PYTHONPATH variable.\n"
 154     msg += "       You need to add <KERNEL_INSTALL_PATH>/bin/salome path.\n"
 155     logging.error(msg)
 156     sys.exit(1)
 157
 158   unittest.main()
 159 #