_once_lock = _allocate_lock()
-def _once(var, initializer):
- """Wrapper to execute an initialization operation just once,
- even if multiple threads reach the same point at the same time.
-
- var is the name (as a string) of the variable to be entered into
- the current global namespace.
-
- initializer is a callable which will return the appropriate initial
- value for variable. It will be called only if variable is not
- present in the global namespace, or its current value is None.
-
- Do not call _once from inside an initializer routine, it will deadlock.
- """
-
- vars = globals()
- # Check first outside the lock.
- if vars.get(var) is not None:
- return
- try:
- _once_lock.acquire()
- # Check again inside the lock.
- if vars.get(var) is not None:
- return
- vars[var] = initializer()
- finally:
- _once_lock.release()
-
class _RandomNameSequence:
"""An instance of _RandomNameSequence generates an endless
sequence of unpredictable strings which can safely be incorporated
def _get_default_tempdir():
"""Calculate the default directory to use for temporary files.
- This routine should be called through '_once' (see above) as we
- do not want multiple threads attempting this calculation simultaneously.
+ This routine should be called exactly once.
We determine whether or not a candidate temp dir is usable by
trying to create and write to a file in that directory. If this
raise IOError, (_errno.ENOENT,
("No usable temporary directory found in %s" % dirlist))
+_name_sequence = None
+
def _get_candidate_names():
"""Common setup sequence for all user-callable interfaces."""
- _once('_name_sequence', _RandomNameSequence)
+ global _name_sequence
+ if _name_sequence is None:
+ _once_lock.acquire()
+ try:
+ if _name_sequence is None:
+ _name_sequence = _RandomNameSequence()
+ finally:
+ _once_lock.release()
return _name_sequence
"""Accessor for tempdir.template."""
return template
+tempdir = None
+
def gettempdir():
"""Accessor for tempdir.tempdir."""
- _once('tempdir', _get_default_tempdir)
+ global tempdir
+ if tempdir is None:
+ _once_lock.acquire()
+ try:
+ if tempdir is None:
+ tempdir = _get_default_tempdir()
+ finally:
+ _once_lock.release()
return tempdir
-def mkstemp(suffix="", prefix=template, dir=gettempdir(), text=False):
+def mkstemp(suffix="", prefix=template, dir=None, text=False):
"""mkstemp([suffix, [prefix, [dir, [text]]]])
User-callable function to create and return a unique temporary
file. The return value is a pair (fd, name) where fd is the
Caller is responsible for deleting the file when done with it.
"""
+ if dir is None:
+ dir = gettempdir()
+
if text:
flags = _text_openflags
else:
return _mkstemp_inner(dir, prefix, suffix, flags)
-def mkdtemp(suffix="", prefix=template, dir=gettempdir()):
+def mkdtemp(suffix="", prefix=template, dir=None):
"""mkdtemp([suffix, [prefix, [dir]]])
User-callable function to create and return a unique temporary
directory. The return value is the pathname of the directory.
Caller is responsible for deleting the directory when done with it.
"""
+ if dir is None:
+ dir = gettempdir()
+
names = _get_candidate_names()
for seq in xrange(TMP_MAX):
raise IOError, (_errno.EEXIST, "No usable temporary directory name found")
-def mktemp(suffix="", prefix=template, dir=gettempdir()):
+def mktemp(suffix="", prefix=template, dir=None):
"""mktemp([suffix, [prefix, [dir]]])
User-callable function to return a unique temporary file name. The
file is not created.
_warn("mktemp is a potential security risk to your program",
RuntimeWarning, stacklevel=2)
+ if dir is None:
+ dir = gettempdir()
+
names = _get_candidate_names()
for seq in xrange(TMP_MAX):
name = names.next()
self.close()
def NamedTemporaryFile(mode='w+b', bufsize=-1, suffix="",
- prefix=template, dir=gettempdir()):
+ prefix=template, dir=None):
"""Create and return a temporary file.
Arguments:
'prefix', 'suffix', 'dir' -- as for mkstemp.
closed.
"""
+ if dir is None:
+ dir = gettempdir()
+
if 'b' in mode:
flags = _bin_openflags
else:
else:
def TemporaryFile(mode='w+b', bufsize=-1, suffix="",
- prefix=template, dir=gettempdir()):
+ prefix=template, dir=None):
"""Create and return a temporary file.
Arguments:
'prefix', 'suffix', 'directory' -- as for mkstemp.
exist when it is closed.
"""
+ if dir is None:
+ dir = gettempdir()
+
if 'b' in mode:
flags = _bin_openflags
else:
test_classes.append(test_exports)
-class test__once(TC):
- """Test the internal function _once."""
-
- def setUp(self):
- tempfile.once_var = None
- self.already_called = 0
-
- def tearDown(self):
- del tempfile.once_var
-
- def callMeOnce(self):
- self.failIf(self.already_called, "callMeOnce called twice")
- self.already_called = 1
- return 24
-
- def do_once(self):
- tempfile._once('once_var', self.callMeOnce)
-
- def test_once_initializes(self):
- """_once initializes its argument"""
-
- self.do_once()
-
- self.assertEqual(tempfile.once_var, 24,
- "once_var=%d, not 24" % tempfile.once_var)
- self.assertEqual(self.already_called, 1,
- "already_called=%d, not 1" % self.already_called)
-
- def test_once_means_once(self):
- """_once calls the callback just once"""
-
- self.do_once()
- self.do_once()
- self.do_once()
- self.do_once()
-
- def test_once_namespace_safe(self):
- """_once does not modify anything but its argument"""
-
- env_copy = tempfile.__dict__.copy()
-
- self.do_once()
-
- env = tempfile.__dict__
-
- a = env.keys()
- a.sort()
- b = env_copy.keys()
- b.sort()
-
- self.failIf(len(a) != len(b))
- for i in xrange(len(a)):
- self.failIf(a[i] != b[i])
-
- key = a[i]
- if key != 'once_var':
- self.failIf(env[key] != env_copy[key])
-
-test_classes.append(test__once)
-
-
class test__RandomNameSequence(TC):
"""Test the internal iterator object _RandomNameSequence."""
projects / python / commitdiff