Commit 0e2bcbd7 authored by Jim Fulton's avatar Jim Fulton

Fixed bug:

- Unix domain sockets didn't work for ZEO (since the addition of IPv6
  support).

https://bugs.launchpad.net/zodb/+bug/663259
parent 305ac133
...@@ -17,6 +17,9 @@ Bugs Fixed ...@@ -17,6 +17,9 @@ Bugs Fixed
https://bugs.launchpad.net/zodb/+bug/665452 https://bugs.launchpad.net/zodb/+bug/665452
- Unix domain sockets didn't work for ZEO (since the addition of IPv6
support). https://bugs.launchpad.net/zodb/+bug/663259
- Removed a missfeature that can cause performance problems when using - Removed a missfeature that can cause performance problems when using
an external garbage collector with ZEO. When objects were deleted an external garbage collector with ZEO. When objects were deleted
from a storage, invalidations were sent to clients. This makes no from a storage, invalidations were sent to clients. This makes no
......
...@@ -33,6 +33,11 @@ class ZEOConfig: ...@@ -33,6 +33,11 @@ class ZEOConfig:
"""Class to generate ZEO configuration file. """ """Class to generate ZEO configuration file. """
def __init__(self, addr): def __init__(self, addr):
if isinstance(addr, str):
self.logpath = addr+'.log'
else:
self.logpath = 'server-%s.log' % addr[1]
addr = '%s:%s' % addr
self.address = addr self.address = addr
self.read_only = None self.read_only = None
self.invalidation_queue_size = None self.invalidation_queue_size = None
...@@ -46,7 +51,7 @@ class ZEOConfig: ...@@ -46,7 +51,7 @@ class ZEOConfig:
def dump(self, f): def dump(self, f):
print >> f, "<zeo>" print >> f, "<zeo>"
print >> f, "address %s:%s" % self.address print >> f, "address " + self.address
if self.read_only is not None: if self.read_only is not None:
print >> f, "read-only", self.read_only and "true" or "false" print >> f, "read-only", self.read_only and "true" or "false"
if self.invalidation_queue_size is not None: if self.invalidation_queue_size is not None:
...@@ -69,10 +74,10 @@ class ZEOConfig: ...@@ -69,10 +74,10 @@ class ZEOConfig:
<eventlog> <eventlog>
level %s level %s
<logfile> <logfile>
path server-%s.log path %s
</logfile> </logfile>
</eventlog> </eventlog>
""" % (self.loglevel, self.address[1]) """ % (self.loglevel, self.logpath)
def __str__(self): def __str__(self):
f = StringIO.StringIO() f = StringIO.StringIO()
...@@ -110,8 +115,15 @@ def start_zeo_server(storage_conf=None, zeo_conf=None, port=None, keep=False, ...@@ -110,8 +115,15 @@ def start_zeo_server(storage_conf=None, zeo_conf=None, port=None, keep=False,
if port is None: if port is None:
raise AssertionError("The port wasn't specified") raise AssertionError("The port wasn't specified")
if isinstance(port, int):
addr = 'localhost', port
adminaddr = 'localhost', port+1
else:
addr = port
adminaddr = port+'-test'
if zeo_conf is None or isinstance(zeo_conf, dict): if zeo_conf is None or isinstance(zeo_conf, dict):
z = ZEOConfig(('localhost', port)) z = ZEOConfig(addr)
if zeo_conf: if zeo_conf:
z.__dict__.update(zeo_conf) z.__dict__.update(zeo_conf)
zeo_conf = z zeo_conf = z
...@@ -149,15 +161,19 @@ def start_zeo_server(storage_conf=None, zeo_conf=None, port=None, keep=False, ...@@ -149,15 +161,19 @@ def start_zeo_server(storage_conf=None, zeo_conf=None, port=None, keep=False,
else: else:
pid = subprocess.Popen(args, env=d, close_fds=True).pid pid = subprocess.Popen(args, env=d, close_fds=True).pid
adminaddr = ('localhost', port + 1)
# We need to wait until the server starts, but not forever. # We need to wait until the server starts, but not forever.
# 30 seconds is a somewhat arbitrary upper bound. A BDBStorage # 30 seconds is a somewhat arbitrary upper bound. A BDBStorage
# takes a long time to open -- more than 10 seconds on occasion. # takes a long time to open -- more than 10 seconds on occasion.
for i in range(120): for i in range(300):
time.sleep(0.25) time.sleep(0.1)
try: try:
if isinstance(adminaddr, str) and not os.path.exists(adminaddr):
continue
logger.debug('connect %s', i) logger.debug('connect %s', i)
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) if isinstance(adminaddr, str):
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
else:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(adminaddr) s.connect(adminaddr)
ack = s.recv(1024) ack = s.recv(1024)
s.close() s.close()
...@@ -170,7 +186,7 @@ def start_zeo_server(storage_conf=None, zeo_conf=None, port=None, keep=False, ...@@ -170,7 +186,7 @@ def start_zeo_server(storage_conf=None, zeo_conf=None, port=None, keep=False,
else: else:
logging.debug('boo hoo') logging.debug('boo hoo')
raise raise
return ('localhost', port), adminaddr, pid, tmpfile return addr, adminaddr, pid, tmpfile
if sys.platform[:3].lower() == "win": if sys.platform[:3].lower() == "win":
...@@ -187,7 +203,10 @@ def shutdown_zeo_server(adminaddr): ...@@ -187,7 +203,10 @@ def shutdown_zeo_server(adminaddr):
# only requires two iterations, but do a third for pure # only requires two iterations, but do a third for pure
# superstition. # superstition.
for i in range(3): for i in range(3):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) if isinstance(adminaddr, str):
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
else:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(.3) s.settimeout(.3)
try: try:
s.connect(adminaddr) s.connect(adminaddr)
......
...@@ -1489,8 +1489,20 @@ def runzeo_logrotate_on_sigusr2(): ...@@ -1489,8 +1489,20 @@ def runzeo_logrotate_on_sigusr2():
>>> _ = p.wait() >>> _ = p.wait()
""" """
def unix_domain_sockets():
"""Make sure unix domain sockets work
>>> addr, _ = start_server(port='./sock')
>>> c = ZEO.connection(addr)
>>> c.root.x = 1
>>> transaction.commit()
>>> c.close()
"""
if sys.platform.startswith('win'): if sys.platform.startswith('win'):
del runzeo_logrotate_on_sigusr2 del runzeo_logrotate_on_sigusr2
del unix_domain_sockets
if sys.version_info >= (2, 6): if sys.version_info >= (2, 6):
import multiprocessing import multiprocessing
......
...@@ -67,7 +67,10 @@ class ZEOTestServer(asyncore.dispatcher): ...@@ -67,7 +67,10 @@ class ZEOTestServer(asyncore.dispatcher):
# Count down to zero, the number of connects # Count down to zero, the number of connects
self._count = 1 self._count = 1
self._label ='%d @ %s' % (os.getpid(), addr) self._label ='%d @ %s' % (os.getpid(), addr)
self.create_socket(socket.AF_INET, socket.SOCK_STREAM) if isinstance(addr, str):
self.create_socket(socket.AF_UNIX, socket.SOCK_STREAM)
else:
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
# Some ZEO tests attempt a quick start of the server using the same # Some ZEO tests attempt a quick start of the server using the same
# port so we have to set the reuse flag. # port so we have to set the reuse flag.
self.set_reuse_addr() self.set_reuse_addr()
...@@ -168,17 +171,15 @@ def main(): ...@@ -168,17 +171,15 @@ def main():
zo = ZEO.runzeo.ZEOOptions() zo = ZEO.runzeo.ZEOOptions()
zo.realize(["-C", configfile]) zo.realize(["-C", configfile])
zeo_port = int(zo.address[1]) addr = zo.address
if zo.auth_protocol == "plaintext": if zo.auth_protocol == "plaintext":
__import__('ZEO.tests.auth_plaintext') __import__('ZEO.tests.auth_plaintext')
# Open the config file and let ZConfig parse the data there. Then remove if isinstance(addr, tuple):
# the config file, otherwise we'll leave turds. test_addr = addr[0], addr[1]+1
# The rest of the args are hostname, portnum else:
test_port = zeo_port + 1 test_addr = addr + '-test'
test_addr = ('localhost', test_port)
addr = ('localhost', zeo_port)
log(label, 'creating the storage server') log(label, 'creating the storage server')
storage = zo.storages[0].open() storage = zo.storages[0].open()
mon_addr = None mon_addr = None
......
...@@ -437,12 +437,13 @@ class ConnectThread(threading.Thread): ...@@ -437,12 +437,13 @@ class ConnectThread(threading.Thread):
return 0 return 0
def _expand_addrlist(self): def _expand_addrlist(self):
for domain, (host, port) in self.addrlist: for domain, addr in self.addrlist:
# AF_INET really means either IPv4 or IPv6, possibly # AF_INET really means either IPv4 or IPv6, possibly
# indirected by DNS. By design, DNS lookup is deferred # indirected by DNS. By design, DNS lookup is deferred
# until connections get established, so that DNS # until connections get established, so that DNS
# reconfiguration can affect failover # reconfiguration can affect failover
if domain == socket.AF_INET: if domain == socket.AF_INET:
host, port = addr
for (family, socktype, proto, cannoname, sockaddr for (family, socktype, proto, cannoname, sockaddr
) in socket.getaddrinfo(host or 'localhost', port): ) in socket.getaddrinfo(host or 'localhost', port):
# for IPv6, drop flowinfo, and restrict addresses # for IPv6, drop flowinfo, and restrict addresses
......
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