Skip to content

N
neoppod
Commit ae37c472 authored by Vincent Pelletier's avatar Vincent Pelletier
Browse files
Options

Use a (non-blocking) lock to prevent two simultaneous reconnection attemps to... 

Use a (non-blocking) lock to prevent two simultaneous reconnection attemps to master instead of a simple variable.


git-svn-id: https://svn.erp5.org/repos/neo/branches/prototype3@266 71dcc9de-d417-0410-9af5-da40c76e7ee4
parent 31481811
Hide whitespace changes
Inline Side-by-side
Showing with 79 additions and 84 deletions
neo/client/dispatcher.py
View file @ ae37c472
......@@ -15,7 +15,7 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
from threading import Thread
from threading import Thread, Lock
from Queue import Empty, Queue
from neo.protocol import PING, Packet, CLIENT_NODE_TYPE, FINISH_TRANSACTION
......@@ -36,7 +36,7 @@ class Dispatcher(Thread):
# and thus redispatch answer to the original thread
self.message_table = {}
# Indicate if we are in process of connection to master node
self.connecting_to_master_node = 0
self.connecting_to_master_node = Lock()
def run(self):
while 1:
......@@ -66,77 +66,80 @@ class Dispatcher(Thread):
This can be called either at bootstrap or when
client got disconnected during process"""
# Indicate we are trying to connect to avoid multiple try a time
self.connecting_to_master_node = 1
from neo.client.handler import ClientEventHandler
if app.pt is not None:
app.pt.clear()
master_index = 0
t = 0
conn = None
# Make application execute remaining message if any
app._waitMessage()
handler = ClientEventHandler(app, app.dispatcher)
while 1:
if t + 1 < time():
if app.pt is not None and app.pt.operational():
# Connected to primary master node and got all informations
break
app.local_var.node_not_ready = 0
if app.primary_master_node is None:
# Try with master node defined in config
try:
addr, port = app.master_node_list[master_index].split(':')
except IndexError:
master_index = 0
addr, port = app.master_node_list[master_index].split(':')
port = int(port)
else:
addr, port = app.primary_master_node.getServer()
# Request Node Identification
conn = MTClientConnection(app.em, handler, (addr, port), connector_handler=connector)
if app.nm.getNodeByServer((addr, port)) is None:
n = MasterNode(server = (addr, port))
app.nm.add(n)
acquired = self.connecting_to_master_node.acquire(blocking=0)
if acquired:
try:
from neo.client.handler import ClientEventHandler
if app.pt is not None:
app.pt.clear()
master_index = 0
t = 0
conn = None
# Make application execute remaining message if any
app._waitMessage()
handler = ClientEventHandler(app, app.dispatcher)
while 1:
if t + 1 < time():
if app.pt is not None and app.pt.operational():
# Connected to primary master node and got all informations
break
app.local_var.node_not_ready = 0
if app.primary_master_node is None:
# Try with master node defined in config
try:
addr, port = app.master_node_list[master_index].split(':')
except IndexError:
master_index = 0
addr, port = app.master_node_list[master_index].split(':')
port = int(port)
else:
addr, port = app.primary_master_node.getServer()
# Request Node Identification
conn = MTClientConnection(app.em, handler, (addr, port), connector_handler=connector)
if app.nm.getNodeByServer((addr, port)) is None:
n = MasterNode(server = (addr, port))
app.nm.add(n)
conn.lock()
try:
msg_id = conn.getNextId()
p = Packet()
p.requestNodeIdentification(msg_id, CLIENT_NODE_TYPE, app.uuid,
'0.0.0.0', 0, app.name)
conn.lock()
try:
msg_id = conn.getNextId()
p = Packet()
p.requestNodeIdentification(msg_id, CLIENT_NODE_TYPE, app.uuid,
'0.0.0.0', 0, app.name)
# Send message
conn.addPacket(p)
conn.expectMessage(msg_id)
self.register(conn, msg_id, app.getQueue())
finally:
conn.unlock()
# Send message
conn.addPacket(p)
conn.expectMessage(msg_id)
self.register(conn, msg_id, app.getQueue())
finally:
conn.unlock()
# Wait for answer
while 1:
try:
self.em.poll(1)
except TypeError:
break
app._waitMessage()
# Now check result
if app.primary_master_node is not None:
if app.primary_master_node == -1:
# Connection failed, try with another master node
app.primary_master_node = None
master_index += 1
break
elif app.primary_master_node.getServer() != (addr, port):
# Master node changed, connect to new one
break
elif app.local_var.node_not_ready:
# Wait a bit and reask again
break
elif app.pt is not None and app.pt.operational():
# Connected to primary master node
break
t = time()
# Wait for answer
while 1:
try:
self.em.poll(1)
except TypeError:
break
app._waitMessage()
# Now check result
if app.primary_master_node is not None:
if app.primary_master_node == -1:
# Connection failed, try with another master node
app.primary_master_node = None
master_index += 1
break
elif app.primary_master_node.getServer() != (addr, port):
# Master node changed, connect to new one
break
elif app.local_var.node_not_ready:
# Wait a bit and reask again
break
elif app.pt is not None and app.pt.operational():
# Connected to primary master node
break
t = time()
logging.info("connected to primary master node %s:%d" %app.primary_master_node.getServer())
app.master_conn = conn
self.connecting_to_master_node = 0
logging.info("connected to primary master node %s:%d" %app.primary_master_node.getServer())
app.master_conn = conn
finally:
self.connecting_to_master_node.release()
neo/client/handler.py
View file @ ae37c472
......@@ -101,9 +101,7 @@ class ClientEventHandler(EventHandler):
elif self.app.primary_master_node is not None and uuid == \
self.app.primary_master_node.getUUID():
logging.critical("connection to primary master node failed")
if self.dispatcher.connecting_to_master_node == 0:
logging.critical("trying reconnection to master node...")
self.dispatcher.connectToPrimaryMasterNode(app, conn)
self.dispatcher.connectToPrimaryMasterNode(app, conn)
else:
# Connection to a storage node failed
node = app.nm.getNodeByServer(conn.getAddress())
......@@ -124,9 +122,7 @@ class ClientEventHandler(EventHandler):
app.master_conn.close()
app.master_conn = None
app.primary_master_node = None
if self.dispatcher.connecting_to_master_node == 0:
logging.critical("trying reconnection to master node...")
self.dispatcher.connectToPrimaryMasterNode(app, conn)
self.dispatcher.connectToPrimaryMasterNode(app, conn)
else:
node = app.nm.getNodeByServer(conn.getAddress())
if isinstance(node, StorageNode):
......@@ -145,9 +141,7 @@ class ClientEventHandler(EventHandler):
app.primary_master_node = -1
if app.master_conn is not None and uuid == app.primary_master_node.getUUID():
logging.critical("connection timeout to primary master node expired")
if self.dispatcher.connecting_to_master_node == 0:
logging.critical("trying reconnection to master node...")
self.dispatcher.connectToPrimaryMasterNode(app, conn)
self.dispatcher.connectToPrimaryMasterNode(app, conn)
else:
node = app.nm.getNodeByServer(conn.getAddress())
if isinstance(node, StorageNode):
......@@ -165,9 +159,7 @@ class ClientEventHandler(EventHandler):
app.primary_master_node = -1
if app.master_conn is not None and uuid == app.primary_master_node.getUUID():
logging.critical("primary master node is broken")
if self.dispatcher.connecting_to_master_node == 0:
logging.critical("trying reconnection to master node...")
self.dispatcher.connectToPrimaryMasterNode(app, conn)
self.dispatcher.connectToPrimaryMasterNode(app, conn)
else:
node = app.nm.getNodeByServer(conn.getAddress())
if isinstance(node, StorageNode):
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7