/usr/local/lib64/python3.6/site-packages/caffe2/python
Edit: /usr/local/lib64/python3.6/site-packages/caffe2/python/net_builder_test.py (11382B)
from caffe2.python import workspace
from caffe2.python.core import Plan, to_execution_step, Net
from caffe2.python.task import Task, TaskGroup, final_output
from caffe2.python.net_builder import ops, NetBuilder
from caffe2.python.session import LocalSession
import unittest
import threading
class PythonOpStats(object):
lock = threading.Lock()
num_instances = 0
num_calls = 0
def python_op_builder():
PythonOpStats.lock.acquire()
PythonOpStats.num_instances += 1
PythonOpStats.lock.release()
def my_op(inputs, outputs):
PythonOpStats.lock.acquire()
PythonOpStats.num_calls += 1
PythonOpStats.lock.release()
return my_op
def _test_loop():
x = ops.Const(5)
y = ops.Const(0)
with ops.loop():
ops.stop_if(ops.EQ([x, ops.Const(0)]))
ops.Add([x, ops.Const(-1)], [x])
ops.Add([y, ops.Const(1)], [y])
return y
def _test_inner_stop(x):
ops.stop_if(ops.LT([x, ops.Const(5)]))
def _test_outer():
x = ops.Const(10)
# test stop_if(False)
with ops.stop_guard() as g1:
_test_inner_stop(x)
# test stop_if(True)
y = ops.Const(3)
with ops.stop_guard() as g2:
_test_inner_stop(y)
# test no stop
with ops.stop_guard() as g4:
ops.Const(0)
# test empty clause
with ops.stop_guard() as g3:
pass
return (
g1.has_stopped(), g2.has_stopped(), g3.has_stopped(), g4.has_stopped())
def _test_if(x):
y = ops.Const(1)
with ops.If(ops.GT([x, ops.Const(50)])):
ops.Const(2, blob_out=y)
with ops.If(ops.LT([x, ops.Const(50)])):
ops.Const(3, blob_out=y)
ops.stop()
ops.Const(4, blob_out=y)
return y
class TestNetBuilder(unittest.TestCase):
def test_ops(self):
with NetBuilder() as nb:
y = _test_loop()
z, w, a, b = _test_outer()
p = _test_if(ops.Const(75))
q = _test_if(ops.Const(25))
plan = Plan('name')
plan.AddStep(to_execution_step(nb))
ws = workspace.C.Workspace()
ws.run(plan)
expected = [
(y, 5),
(z, False),
(w, True),
(a, False),
(b, False),
(p, 2),
(q, 3),
]
for b, expected in expected:
actual = ws.blobs[str(b)].fetch()
self.assertEquals(actual, expected)
def _expected_loop(self):
total = 0
total_large = 0
total_small = 0
total_tiny = 0
for loop_iter in range(10):
outer = loop_iter * 10
for inner_iter in range(loop_iter):
val = outer + inner_iter
if val >= 80:
total_large += val
elif val >= 50:
total_small += val
else:
total_tiny += val
total += val
return total, total_large, total_small, total_tiny
def _actual_loop(self):
total = ops.Const(0)
total_large = ops.Const(0)
total_small = ops.Const(0)
total_tiny = ops.Const(0)
with ops.loop(10) as loop:
outer = ops.Mul([loop.iter(), ops.Const(10)])
with ops.loop(loop.iter()) as inner:
val = ops.Add([outer, inner.iter()])
with ops.If(ops.GE([val, ops.Const(80)])) as c:
ops.Add([total_large, val], [total_large])
with c.Elif(ops.GE([val, ops.Const(50)])) as c:
ops.Add([total_small, val], [total_small])
with c.Else():
ops.Add([total_tiny, val], [total_tiny])
ops.Add([total, val], total)
return [
final_output(x)
for x in [total, total_large, total_small, total_tiny]
]
def test_net_multi_use(self):
with Task() as task:
total = ops.Const(0)
net = Net('my_net')
net.Add([total, net.Const(1)], [total])
ops.net(net)
ops.net(net)
result = final_output(total)
with LocalSession() as session:
session.run(task)
self.assertEquals(2, result.fetch())
def test_loops(self):
with Task() as task:
out_actual = self._actual_loop()
with LocalSession() as session:
session.run(task)
expected = self._expected_loop()
actual = [o.fetch() for o in out_actual]
for e, a in zip(expected, actual):
self.assertEquals(e, a)
def test_setup(self):
with Task() as task:
with ops.task_init():
one = ops.Const(1)
two = ops.Add([one, one])
with ops.task_init():
three = ops.Const(3)
accum = ops.Add([two, three])
# here, accum should be 5
with ops.task_exit():
# here, accum should be 6, since this executes after lines below
seven_1 = ops.Add([accum, one])
six = ops.Add([accum, one])
ops.Add([accum, one], [accum])
seven_2 = ops.Add([accum, one])
o6 = final_output(six)
o7_1 = final_output(seven_1)
o7_2 = final_output(seven_2)
with LocalSession() as session:
session.run(task)
self.assertEquals(o6.fetch(), 6)
self.assertEquals(o7_1.fetch(), 7)
self.assertEquals(o7_2.fetch(), 7)
def test_multi_instance_python_op(self):
"""
When task instances are created at runtime, C++ concurrently creates
multiple instances of operators in C++, and concurrently destroys them
once the task is finished. This means that the destructor of PythonOp
will be called concurrently, so the GIL must be acquired. This
test exercises this condition.
"""
with Task(num_instances=64) as task:
with ops.loop(4):
ops.Python((python_op_builder, [], {}))([], [])
with LocalSession() as session:
PythonOpStats.num_instances = 0
PythonOpStats.num_calls = 0
session.run(task)
self.assertEquals(PythonOpStats.num_instances, 64)
self.assertEquals(PythonOpStats.num_calls, 256)
def test_multi_instance(self):
NUM_INSTANCES = 10
NUM_ITERS = 15
with TaskGroup() as tg:
with Task(num_instances=NUM_INSTANCES):
with ops.task_init():
counter1 = ops.CreateCounter([], ['global_counter'])
counter2 = ops.CreateCounter([], ['global_counter2'])
counter3 = ops.CreateCounter([], ['global_counter3'])
# both task_counter and local_counter should be thread local
with ops.task_instance_init():
task_counter = ops.CreateCounter([], ['task_counter'])
local_counter = ops.CreateCounter([], ['local_counter'])
with ops.loop(NUM_ITERS):
ops.CountUp(counter1)
ops.CountUp(task_counter)
ops.CountUp(local_counter)
# gather sum of squares of local counters to make sure that
# each local counter counted exactly up to NUM_ITERS, and
# that there was no false sharing of counter instances.
with ops.task_instance_exit():
count2 = ops.RetrieveCount(task_counter)
with ops.loop(ops.Mul([count2, count2])):
ops.CountUp(counter2)
# This should have the same effect as the above
count3 = ops.RetrieveCount(local_counter)
with ops.loop(ops.Mul([count3, count3])):
ops.CountUp(counter3)
# The code below will only run once
with ops.task_exit():
total1 = final_output(ops.RetrieveCount(counter1))
total2 = final_output(ops.RetrieveCount(counter2))
total3 = final_output(ops.RetrieveCount(counter3))
with LocalSession() as session:
session.run(tg)
self.assertEquals(total1.fetch(), NUM_INSTANCES * NUM_ITERS)
self.assertEquals(total2.fetch(), NUM_INSTANCES * (NUM_ITERS ** 2))
self.assertEquals(total3.fetch(), NUM_INSTANCES * (NUM_ITERS ** 2))
def test_if_net(self):
with NetBuilder() as nb:
x0 = ops.Const(0)
x1 = ops.Const(1)
x2 = ops.Const(2)
y0 = ops.Const(0)
y1 = ops.Const(1)
y2 = ops.Const(2)
# basic logic
first_res = ops.Const(0)
with ops.IfNet(ops.Const(True)):
ops.Const(1, blob_out=first_res)
with ops.Else():
ops.Const(2, blob_out=first_res)
second_res = ops.Const(0)
with ops.IfNet(ops.Const(False)):
ops.Const(1, blob_out=second_res)
with ops.Else():
ops.Const(2, blob_out=second_res)
# nested and sequential ifs,
# empty then/else,
# passing outer blobs into branches,
# writing into outer blobs, incl. into input blob
# using local blobs
with ops.IfNet(ops.LT([x0, x1])):
local_blob = ops.Const(900)
ops.Add([ops.Const(100), local_blob], [y0])
gt = ops.GT([x1, x2])
with ops.IfNet(gt):
# empty then
pass
with ops.Else():
ops.Add([y1, local_blob], [local_blob])
ops.Add([ops.Const(100), y1], [y1])
with ops.IfNet(ops.EQ([local_blob, ops.Const(901)])):
ops.Const(7, blob_out=y2)
ops.Add([y1, y2], [y2])
with ops.Else():
# empty else
pass
plan = Plan('if_net_test')
plan.AddStep(to_execution_step(nb))
ws = workspace.C.Workspace()
ws.run(plan)
first_res_value = ws.blobs[str(first_res)].fetch()
second_res_value = ws.blobs[str(second_res)].fetch()
y0_value = ws.blobs[str(y0)].fetch()
y1_value = ws.blobs[str(y1)].fetch()
y2_value = ws.blobs[str(y2)].fetch()
self.assertEquals(first_res_value, 1)
self.assertEquals(second_res_value, 2)
self.assertEquals(y0_value, 1000)
self.assertEquals(y1_value, 101)
self.assertEquals(y2_value, 108)
self.assertTrue(str(local_blob) not in ws.blobs)
def test_while_net(self):
with NetBuilder() as nb:
x = ops.Const(0)
y = ops.Const(0)
with ops.WhileNet():
with ops.Condition():
ops.Add([x, ops.Const(1)], [x])
ops.LT([x, ops.Const(7)])
ops.Add([x, y], [y])
plan = Plan('while_net_test')
plan.AddStep(to_execution_step(nb))
ws = workspace.C.Workspace()
ws.run(plan)
x_value = ws.blobs[str(x)].fetch()
y_value = ws.blobs[str(y)].fetch()
self.assertEqual(x_value, 7)
self.assertEqual(y_value, 21)