<s> import _multiprocessing <EOL> import sys <EOL> import os <EOL> import socket <EOL> sys . path . append ( os . getcwd ( ) ) <EOL> from numpush . posix_io . splice import posix_splice , posix_splice_thread , posix_splice_sockets , SPLICE_MORE , SPLICE_MOVE <EOL> DATA = '<STR_LIT:a>' * <NUM_LIT> <EOL> def test_single ( ) : <EOL> os . mkfifo ( '<STR_LIT:w>' ) <EOL> os . mkfifo ( '<STR_LIT:r>' ) <EOL> try : <EOL> w = open ( '<STR_LIT:w>' , '<STR_LIT>' ) <EOL> r = open ( '<STR_LIT:r>' , '<STR_LIT>' ) <EOL> w . write ( DATA ) <EOL> w . flush ( ) <EOL> flags = SPLICE_MORE | SPLICE_MOVE <EOL> posix_splice ( w , r , <NUM_LIT> , flags ) <EOL> assert r . read ( <NUM_LIT> ) == DATA <EOL> finally : <EOL> os . unlink ( '<STR_LIT:w>' ) <EOL> os . unlink ( '<STR_LIT:r>' ) <EOL> def test_multithreaded ( ) : <EOL> os . mkfifo ( '<STR_LIT:w>' ) <EOL> os . mkfifo ( '<STR_LIT:r>' ) <EOL> try : <EOL> w = open ( '<STR_LIT:w>' , '<STR_LIT>' ) <EOL> r = open ( '<STR_LIT:r>' , '<STR_LIT>' ) <EOL> w . write ( DATA ) <EOL> w . flush ( ) <EOL> flags = SPLICE_MORE | SPLICE_MOVE <EOL> thread = posix_splice_thread ( w , r , <NUM_LIT> , flags ) <EOL> thread . join ( ) <EOL> assert r . read ( <NUM_LIT> ) == DATA <EOL> finally : <EOL> os . unlink ( '<STR_LIT:w>' ) <EOL> os . unlink ( '<STR_LIT:r>' ) </s>
