FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
cpython/Lib/concurrent/interpreters/_crossinterp.py at 3.15 · python/cpython · GitHub
python
cpython
Repository navigation
Code
Issues
5k+
(5k+)
Pull requests
2.3k
(2.3k)
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
cpython
/
Lib
/
concurrent
/
interpreters
/
_crossinterp.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
107 lines (84 loc) · 2.84 KB
Breadcrumbs
cpython
/
Lib
/
concurrent
/
interpreters
/
_crossinterp.py
Copy path
File metadata and controls
107 lines (84 loc) · 2.84 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
"""Common code between queues and channels."""
class
ItemInterpreterDestroyed
(
Exception
):
"""Raised when trying to get an item whose interpreter was destroyed."""
class
classonly
:
"""A non-data descriptor that makes a value only visible on the class.
This is like the "classmethod" builtin, but does not show up on
instances of the class. It may be used as a decorator.
"""
def
__init__
(
self
,
value
):
self
.
value
=
value
self
.
getter
=
classmethod
(
value
).
__get__
self
.
name
=
None
def
__set_name__
(
self
,
cls
,
name
):
if
self
.
name
is
not
None
:
raise
TypeError
(
'already used'
)
self
.
name
=
name
def
__get__
(
self
,
obj
,
cls
):
if
obj
is
not
None
:
raise
AttributeError
(
self
.
name
)
# called on the class
return
self
.
getter
(
None
,
cls
)
class
UnboundItem
:
"""Represents a cross-interpreter item no longer bound to an interpreter.
An item is unbound when the interpreter that added it to the
cross-interpreter container is destroyed.
"""
__slots__
=
()
@
classonly
def
singleton
(
cls
,
kind
,
module
,
name
=
'UNBOUND'
):
doc
=
cls
.
__doc__
if
doc
:
doc
=
doc
.
replace
(
'cross-interpreter container'
,
kind
,
).
replace
(
'cross-interpreter'
,
kind
,
)
subclass
=
type
(
f'Unbound
{
kind
.
capitalize
()
}
Item'
,
(
cls
,),
{
"_MODULE"
:
module
,
"_NAME"
:
name
,
"__doc__"
:
doc
,
},
)
return
object
.
__new__
(
subclass
)
_MODULE
=
__name__
_NAME
=
'UNBOUND'
def
__new__
(
cls
):
raise
Exception
(
f'use
{
cls
.
_MODULE
}
.
{
cls
.
_NAME
}
'
)
def
__repr__
(
self
):
return
f'
{
self
.
_MODULE
}
.
{
self
.
_NAME
}
'
# return f'interpreters._queues.UNBOUND'
UNBOUND
=
object
.
__new__
(
UnboundItem
)
UNBOUND_ERROR
=
object
()
UNBOUND_REMOVE
=
object
()
_UNBOUND_CONSTANT_TO_FLAG
=
{
UNBOUND_REMOVE
:
1
,
UNBOUND_ERROR
:
2
,
UNBOUND
:
3
,
}
_UNBOUND_FLAG_TO_CONSTANT
=
{
v
:
k
for
k
,
v
in
_UNBOUND_CONSTANT_TO_FLAG
.
items
()}
def
serialize_unbound
(
unbound
):
op
=
unbound
try
:
flag
=
_UNBOUND_CONSTANT_TO_FLAG
[
op
]
except
KeyError
:
raise
NotImplementedError
(
f'unsupported unbound replacement op
{
op
!r
}
'
)
return
flag
,
def
resolve_unbound
(
flag
,
exctype_destroyed
):
try
:
op
=
_UNBOUND_FLAG_TO_CONSTANT
[
flag
]
except
KeyError
:
raise
NotImplementedError
(
f'unsupported unbound replacement op
{
flag
!r
}
'
)
if
op
is
UNBOUND_REMOVE
:
# "remove" not possible here
raise
NotImplementedError
elif
op
is
UNBOUND_ERROR
:
raise
exctype_destroyed
(
"item's original interpreter destroyed"
)
elif
op
is
UNBOUND
:
return
UNBOUND
else
:
raise
NotImplementedError
(
repr
(
op
))
Back
|
FazBrowse Home
|
New Git URL