2 # Copyright (C) 2013-2024 CEA, EDF, OPEN CASCADE
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.
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.
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
18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
23 import multiprocessing
28 def port_reservation(obtained_ports, preferred=None, name = None, messages=None, expected=None):
29 from PortManager import getPort
31 port = getPort(preferred)
34 print("obtained port = %s"%port)
36 obtained_ports.put(port)
38 if expected and messages:
39 _name = name + " : " if name else ""
41 messages.put(_name + "used port= %s, expected port = %s"%(port, expected))
43 messages.put(_name + "OK")
47 class TestMinimalExample(unittest.TestCase):
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()
57 multiprocessing.Process(target=port_reservation, args=(obtained_ports,))
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",
74 # Try to get specific port number
75 p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))
79 # Try to get specific port number
80 p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2899, "testSequential 2899:1",
86 print ("release port 2899")
87 p = multiprocessing.Process(target=releasePort, args=(2899,))
91 # Try to get specific port number
92 p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2899, "testSequential 2899:2",
98 while not messages.empty():
99 message = messages.get()
100 if "OK" not in message:
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,))
112 print("END testSequential")
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()
122 multiprocessing.Process(target=port_reservation, args=(obtained_ports,))
127 # Try to get specific port number
128 p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2872, "testSequential 2872",
132 # Try to get specific port number
133 p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))
136 # Try to get specific port number
137 p = multiprocessing.Process(target=port_reservation, args=(obtained_ports, 2812,))
147 while not messages.empty():
148 message = messages.get()
149 if "OK" not in message:
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,))
161 print("END testConcurrent")
165 if __name__ == "__main__":
166 omniorb_user_path = os.getenv("OMNIORB_USER_PATH")
167 if not omniorb_user_path:
169 msg += "Error: please set OMNIORB_USER_PATH variable.\n"
170 msg += " Usually this points to your application USERS directory.\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"