From 9a91b6f95d48997b9bf5f4460218b45d05da544b Mon Sep 17 00:00:00 2001 From: fawce Date: Wed, 18 Jan 2012 21:18:04 -0500 Subject: [PATCH] added a test package, converted the hacky test code into cases, fixed bunch off issues with transforms and merge. --- backtest/util.py | 30 ++++++++- data/feed.py | 46 ++++++++----- data/sources/equity.py | 116 +++++++++++++++++++------------- data/transforms.py | 146 ++++++++++++++++++++++------------------- qbt_client.py | 40 +++++------ test.py | 82 ++--------------------- test/__init__.py | 0 test/testmessaging.py | 66 +++++++++++++++++++ 8 files changed, 295 insertions(+), 231 deletions(-) create mode 100644 test/__init__.py create mode 100644 test/testmessaging.py diff --git a/backtest/util.py b/backtest/util.py index 082b1ed3..8d874238 100644 --- a/backtest/util.py +++ b/backtest/util.py @@ -5,6 +5,9 @@ Small classes to assist with db access, timezone calculations, and so on. import datetime import pytz import json +import logging +import uuid +import zmq class DocWrap(): """ @@ -114,7 +117,7 @@ class ParallelBuffer(object): class MergedParallelBuffer(ParallelBuffer): - def merge_next(self): + def next(self): if(not(self.is_full() or self.draining)): return @@ -123,4 +126,27 @@ class MergedParallelBuffer(ParallelBuffer): if(len(events) > 0): result[source] = events.pop(0) - return result \ No newline at end of file + return result + + +class FeedSync(object): + + def __init__(self, feed, name): + self.feed = feed + self.id = "{name}-{id}".format(name=name, id=uuid.uuid1()) + self.feed.register_sync(self.id) + self.logger = logging.getLogger() + #self.logger.info("registered {id} with feed".format(id=self.id)) + + def confirm(self): + context = zmq.Context() + #synchronize with feed + sync_socket = context.socket(zmq.REQ) + sync_socket.connect(self.feed.sync_address) + # send a synchronization request to the feed + sync_socket.send(self.id) + # wait for synchronization reply from the feed + sync_socket.recv() + sync_socket.close() + context.term() + self.logger.info("sync'd feed from {id}".format(id = self.id)) \ No newline at end of file diff --git a/data/feed.py b/data/feed.py index 5e3847a1..e1962dfb 100644 --- a/data/feed.py +++ b/data/feed.py @@ -6,24 +6,27 @@ import logging class DataFeed(object): - def __init__(self, db, subscriber_count): + def __init__(self, config): self.logger = logging.getLogger() self.data_address = "tcp://127.0.0.1:{port}".format(port=10101) self.sync_address = "tcp://127.0.0.1:{port}".format(port=10102) self.feed_address = "tcp://127.0.0.1:{port}".format(port=10103) - self.db = db + self.client_register = {} + self.data_workers = {} - emt1 = EquityMinuteTrades(133, self.db, self.data_address, self.sync_address, 1) - self.data_workers[1] = emt1 - emt2 = EquityMinuteTrades(134, self.db, self.data_address, self.sync_address, 2) - self.data_workers[2] = emt2 - + self.config = config + for name, info in config.iteritems(): + if(info['class'] == "EquityMinuteTrades"): + emt = EquityMinuteTrades(info['sid'], self, name) + self.data_workers[name] = emt + elif(info['class'] == "RandomEquityTrades"): + ret = RandomEquityTrades(info['sid'], self, name, info['count']) + self.data_workers[name] = ret + self.data_buffer = ParallelBuffer(self.data_workers.keys()) - self.sync_count = subscriber_count + len(self.data_workers) - - + def start_data_workers(self): """Start a sub-process for each datasource.""" for source_id, source in self.data_workers.iteritems(): @@ -31,21 +34,29 @@ class DataFeed(object): source.start() self.logger.info("ds processes launched") + def register_sync(self, sync_id): + self.client_register[sync_id] = "UNCONFIRMED" + + def registration_complete(self): + for sync_id, status in self.client_register.iteritems(): + if status == "UNCONFIRMED": + return False + + return True + def sync_clients(self): # Socket to receive signals self.logger.info("waiting for all datasources and clients to be ready") self.syncservice = self.context.socket(zmq.REP) self.syncservice.bind(self.sync_address) - - subscribers = 1 - while subscribers <= self.sync_count: - self.logger.info("sync'ing {count} of {total}".format(count=subscribers, total=self.sync_count)) + while not self.registration_complete(): # wait for synchronization request msg = self.syncservice.recv() + self.client_register[msg] = "CONFIRMED" + #self.logger.info("confirmed {id}".format(id=msg)) # send synchronization reply - self.syncservice.send('') - subscribers += 1 + self.syncservice.send('CONFIRMED') self.syncservice.close() self.logger.info("sync'd all datasources and clients") @@ -62,7 +73,7 @@ class DataFeed(object): self.data_socket.bind(self.data_address) #create the feed - self.feed_socket = self.context.socket(zmq.PUSH) + self.feed_socket = self.context.socket(zmq.PUB) self.feed_socket.bind(self.feed_address) self.data_buffer.out_socket = self.feed_socket @@ -92,7 +103,6 @@ class DataFeed(object): #send the DONE message self.feed_socket.send("DONE") - self.logger.info("received {n} messages, sent {m} messages".format(n=self.data_buffer.received_count, m=self.data_buffer.sent_count)) self.data_socket.close() self.feed_socket.close() diff --git a/data/sources/equity.py b/data/sources/equity.py index 90962e34..28021268 100644 --- a/data/sources/equity.py +++ b/data/sources/equity.py @@ -7,65 +7,53 @@ import pytz import copy import multiprocessing import logging +import random from pymongo import ASCENDING, DESCENDING from backtest.util import * +from qbt_server import connect_db -class EquityMinuteTrades(object): - - def __init__(self, sid, db, data_address, sync_address, source_id): - self.sid = sid - self.db = db - self.source_id = source_id - self.logger = logging.getLogger() - self.sync_address = sync_address - self.data_address = data_address - self.logger.info("data address is {ds}".format(ds=data_address)) - +class DataSource(object): + def __init__(self, feed, source_id): + self.source_id = source_id + self.logger = logging.getLogger() + self.feed = feed + self.sync = FeedSync(self.feed, str(source_id)) + self.data_address = self.feed.data_address + self.logger.info("data address is {ds}".format(ds=self.feed.data_address)) self.cur_event = None - + def start(self): self.proc = multiprocessing.Process(target=self.run) self.proc.start() - - def run(self): - self.logger.info("starting data source:{sid} on {addr}".format(sid=self.sid, addr=self.data_address)) - self.context = zmq.Context() - #synchronize with feed - sync_socket = self.context.socket(zmq.REQ) - sync_socket.connect(self.sync_address) - # send a synchronization request to the feed - sync_socket.send('') - # wait for synchronization reply from the feed - sync_socket.recv() - sync_socket.close() + + def open(self): + self.logger.info("starting data source:{source_id} on {addr}".format(source_id=self.source_id, addr=self.feed.data_address)) + self.context = zmq.Context() #create the data sink. Based on http://zguide.zeromq.org/py:tasksink2 self.data_socket = self.context.socket(zmq.PUSH) self.data_socket.connect(self.data_address) - eventQS = self.db.equity.trades.minute.find(fields=["sid","price","volume","dt"], - spec={"sid":self.sid}, - sort=[("dt",ASCENDING)], - slave_ok=True) - self.logger.info("found {count} events".format(count=eventQS.count())) - - - for doc in eventQS: - doc_dt = doc['dt'].replace(tzinfo = pytz.utc) - doc_dt_str = format_date(doc_dt) - event = copy.copy(doc) - event['dt'] = doc_dt_str - event['s'] = self.source_id - event['type'] = 'event' - del(event['_id']) + #signal we are ready + self.sync.confirm() + + def run(self): + try: + self.open() + self.send_all() + self.close() + except Exception as err: + self.logger.error(err, "Unexpected failure running datasource - {name}.".format(name=name)) - #send this event to feed address - #self.logger.info("sending {event}".format(event=event)) - self.data_socket.send(json.dumps(event)) - + def send(self, event): + event['s'] = self.source_id + event['type'] = 'event' + self.data_socket.send(json.dumps(event)) + + def close(self): done_msg = {} done_msg['type'] = 'DONE' done_msg['s'] = self.source_id @@ -73,9 +61,49 @@ class EquityMinuteTrades(object): self.data_socket.close() self.context.term() self.logger.info("finished processing data source") + +class EquityMinuteTrades(DataSource): + + def __init__(self, sid, feed, source_id): + self.sid = sid + self.connection, self.db = connect_db() + DataSource.__init__(self, feed, source_id) + + + def send_all(self): + eventQS = self.db.equity.trades.minute.find(fields=["sid","price","volume","dt"], + spec={"sid":self.sid}, + sort=[("dt",ASCENDING)], + slave_ok=True) + self.logger.info("found {count} events".format(count=eventQS.count())) - + for doc in eventQS: + doc_dt = doc['dt'].replace(tzinfo = pytz.utc) + doc_dt_str = format_date(doc_dt) + event = copy.copy(doc) + event['dt'] = doc_dt_str + del(event['_id']) + self.send(event) + + +class RandomEquityTrades(DataSource): + + def __init__(self, sid, feed, source_id, count): + DataSource.__init__(self, feed, source_id) + self.count = count + self.sid = sid + + def send_all(self): + trade_start = datetime.datetime.now() + minute = datetime.timedelta(minutes=1) + price = random.uniform(5.0,50.0) + + for i in range(self.count): + price = price + random.uniform(-0.05,0.05) + event = {'sid':self.sid, 'dt':format_date(trade_start + (minute * i)),'price':price, 'volume':random.randrange(100,10000,100)} + self.send(event) + diff --git a/data/transforms.py b/data/transforms.py index 919b427f..da509440 100644 --- a/data/transforms.py +++ b/data/transforms.py @@ -3,13 +3,14 @@ import logging import datetime import json import config +import copy import multiprocessing from backtest.util import * class Transform(object): """Parent class for feed transforms. Subclass to create a new derived value from the combined feed.""" - def __init__(self, feed_address, result_address, sync_address, config_dict): + def __init__(self, feed, config_dict, result_address): """ feed_address - zmq socket address, Transform will CONNECT a PULL socket and receive messages until "DONE" is received. result_address - zmq socket address, Transform will CONNECT a PUSH socket and send messaes until feed_socket receives "DONE" @@ -19,42 +20,38 @@ class Transform(object): the last transform in a series should be server=True so that clients can connect. """ self.logger = logging.getLogger() - self.feed_address = feed_address + self.feed = feed + self.feed_address = feed.feed_address self.result_address = result_address - self.sync_address = sync_address self.config = config.Config(config_dict) self.name = self.config.get_string('name') + self.sync = FeedSync(feed, self.name) self.state = {} self.state['name'] = self.name self.received_count = 0 self.sent_count = 0 - + def run(self): + self.open() + self.process_all() + self.close() + + def open(self): self.context = zmq.Context() self.logger.info("starting {name} transform".format(name = self.name)) - #create the feed PULL. - self.feed_socket = self.context.socket(zmq.PULL) + #create the feed SUB. + self.feed_socket = self.context.socket(zmq.SUB) self.feed_socket.connect(self.feed_address) + self.feed_socket.setsockopt(zmq.SUBSCRIBE,'') #create the result PUSH self.result_socket = self.context.socket(zmq.PUSH) self.result_socket.connect(self.result_address) - self.logger.info("sync'ing feed from {name}".format(name = self.name)) - #synchronize with feed - sync_socket = self.context.socket(zmq.REQ) - sync_socket.connect(self.sync_address) - # send a synchronization request to the feed - sync_socket.send('') - # wait for synchronization reply from the feed - sync_socket.recv() - sync_socket.close() - + def process_all(self): self.logger.info("starting {name} event loop".format(name = self.name)) - self.run_loop() - - def run_loop(self): + self.sync.confirm() while True: message = self.feed_socket.recv() @@ -65,9 +62,11 @@ class Transform(object): event = json.loads(message) cur_state = self.update(event) cur_state['dt'] = event['dt'] + cur_state['name'] = self.name self.result_socket.send(json.dumps(cur_state)) self.sent_count += 1 - + + def close(self): self.logger.info("Transform {name} recieved {r} and sent {s}".format(name=self.name, r=self.received_count, s=self.sent_count)) self.feed_socket.close() @@ -80,8 +79,8 @@ class Transform(object): class MovingAverage(Transform): - def __init__(self, feed_address, result_address, sync_address, props, server=False): - Transform.__init__(self, feed_address, result_address, sync_address, props) + def __init__(self, feed, props, result_address="tcp://127.0.0.1:20202"): + Transform.__init__(self, feed, props, result_address) self.events = [] self.window = datetime.timedelta(days = self.config.get_integer('days'), @@ -119,15 +118,15 @@ class MergedTransformsFeed(Transform): """ Merge data feed and array of transform feeds into a single result vector. PULL from feed PULL from child transforms - PUSH to client + PUSH merged message to client """ - def __init__(self, feed_address, result_address, sync_address, props): + def __init__(self, feed, props): """ config - must have an entry for 'transforms':array of dicts, which are convertedto configs. """ - Transform.__init__(self, feed_address, result_address, sync_address, props) + Transform.__init__(self, feed, props, "tcp://127.0.0.1:20202") self.transform_address = "tcp://127.0.0.1:{port}".format(port=10104) self.transform_socket = None self.create_transforms(self.config.transforms) @@ -138,10 +137,12 @@ class MergedTransformsFeed(Transform): for props in configs: class_name = props['class'] if(class_name == 'MovingAverage'): - mavg = MovingAverage(self.feed_address, self.transform_address, self.sync_address, props) + mavg = MovingAverage(self.feed, props, self.transform_address) self.transforms[mavg.name] = mavg - self.data_buffer = ParallelBuffer(self.transforms.keys()) + keys = copy.copy(self.transforms.keys()) + keys.append("feed") #for the raw feed + self.data_buffer = MergedParallelBuffer(keys) for name, transform in self.transforms.iteritems(): self.logger.info("starting {name}".format(name=name)) @@ -149,51 +150,62 @@ class MergedTransformsFeed(Transform): proc.start() self.buffers = {} - for name, transform in self.transforms: + for name, transform in self.transforms.iteritems(): self.buffers[name] = [] - def get_socket(self): + def open(self): + self.context = zmq.Context() - if(self.transform_socket == None): - #create the feed PULL. - self.transform_socket = self.context.socket(zmq.PULL) - self.transform_socket.bind(self.transform_address) - return self.transform_socket - - def run_loop(self): + self.logger.info("starting {name} transform".format(name = self.name)) + #create the feed SUB. + self.feed_socket = self.context.socket(zmq.SUB) + self.feed_socket.connect(self.feed_address) + self.feed_socket.setsockopt(zmq.SUBSCRIBE,'') + + #create the result PUSH + self.result_socket = self.context.socket(zmq.PUSH) + self.result_socket.bind(self.result_address) + + #create the transform PULL. + self.transform_socket = self.context.socket(zmq.PULL) + self.transform_socket.bind(self.transform_address) + self.data_buffer.out_socket = self.result_socket + + # Initialize poll set + self.poller = zmq.Poller() + self.poller.register(self.feed_socket, zmq.POLLIN) + self.poller.register(self.transform_socket, zmq.POLLIN) + + def close(self): + self.transform_socket.close() + Transform.close(self) + + def process_all(self): + self.sync.confirm() while True: - #get original feed message - message = self.feed_socket.recv() - self.received_count += 1 - if(message == "DONE"): - self.result_socket.send("DONE") - break; - event = json.loads(message) - self.data_buffer.append(event['name'], event) - merged_event = self.data_buffer.merge_next() - if(merged_event != None): - self.result_socket.send(json.dumps(merged_event)) - self.sent_count += 1 - - self.logger.info("Transform {name} recieved {r} and sent {s}".format(name=self.name, r=self.received_count, s=self.sent_count)) + socks = dict(self.poller.poll()) - self.feed_socket.close() - self.result_socket.close() - self.context.term() - - def update(self, event): - - state = {} - state['feed'] = event - - count = 0 - while count < len(self.transforms): - message = self.get_socket().recv() - if(message == "DONE"): - return "DONE" - data = json.loads(message) - state[data['name']] = data + if self.feed_socket in socks and socks[self.feed_socket] == zmq.POLLIN: + message = self.feed_socket.recv() + self.received_count += 1 + if(message != "DONE"): + event = json.loads(message) + self.data_buffer.append("feed",event) - return state + if self.transform_socket in socks and socks[self.transform_socket] == zmq.POLLIN: + t_message = self.transform_socket.recv() + if(message != "DONE"): + t_event = json.loads(t_message) + self.data_buffer.append(t_event['name'], t_event) + + + self.data_buffer.send_next() + + #drain any remaining messages in the buffer + self.data_buffer.drain() + + #signal to client that we're done + self.result_socket.send("DONE") + self.logger.info("Transform {name} recieved {r} and sent {s}".format(name=self.name, r=self.data_buffer.received_count, s=self.data_buffer.sent_count)) \ No newline at end of file diff --git a/qbt_client.py b/qbt_client.py index 1af88ecc..097dfaa9 100644 --- a/qbt_client.py +++ b/qbt_client.py @@ -8,11 +8,13 @@ from backtest.util import * class TestClient(object): - def __init__(self,feed_address, sync_address, bind=False): - self.logger = logging.getLogger() - self.feed_address = feed_address - self.sync_address = sync_address - self.bind = bind + def __init__(self,feed, address, bind=False): + self.logger = logging.getLogger() + self.feed = feed + self.address = address + self.sync = FeedSync(feed, "testclient") + self.bind = bind + self.received_count = 0 def run(self): @@ -22,41 +24,33 @@ class TestClient(object): self.data_feed = self.context.socket(zmq.PULL) if(self.bind): - self.logger.info("binding to {feed_address}".format(feed_address=self.feed_address)) - self.data_feed.bind(self.feed_address) + self.logger.info("binding to {address}".format(address=self.address)) + self.data_feed.bind(self.address) else: - self.data_feed.connect(self.feed_address) + self.logger.info("connecting to {address}".format(address=self.address)) + self.data_feed.connect(self.address) - self.logger.info("synchronizing with data feed") - - #synchronize with feed - sync_socket = self.context.socket(zmq.REQ) - sync_socket.connect(self.sync_address) - # send a synchronization request to the feed - sync_socket.send('') - # wait for synchronization reply from the feed - sync_socket.recv() - sync_socket.close() + self.sync.confirm() self.logger.info("Starting the client loop") - counter = 0 prev_dt = None while True: msg = self.data_feed.recv() - counter += 1 if(msg == "DONE"): self.logger.info("DONE!") break + self.received_count += 1 + self.logger.info(msg) event = json.loads(msg) if(prev_dt != None): if(not event['dt'] >= prev_dt): raise Exception("message arrived out of order: {date} after {prev}".format(date=event['dt'], prev=prev_dt)) prev_dt = event['dt'] - if(counter % 100 == 0): - self.logger.info("received {n} messages".format(n=counter)) + if(self.received_count % 100 == 0): + self.logger.info("received {n} messages".format(n=self.received_count)) - self.logger.info("received {n} messages".format(n=counter)) + self.logger.info("received {n} messages".format(n=self.received_count)) self.data_feed.close() self.context.term() \ No newline at end of file diff --git a/test.py b/test.py index c1d91cb7..5fd54f4b 100644 --- a/test.py +++ b/test.py @@ -1,78 +1,6 @@ -import zmq -import logging -import tornado -from data.sources.equity import * -from data.feed import * -from data.transforms import MergedTransformsFeed, MovingAverage +import unittest +from test.testmessaging import * -from qbt_server import connect_db -from qbt_client import TestClient - -def datafeed(): - connection, db = connect_db() - logger = logging.getLogger() - feed = DataFeed(db, 2) - #feed.run() - feed_proc = multiprocessing.Process(target=feed.run) - feed_proc.start() - - config = {} - config['name'] = '**merged feed**' - config['transforms'] = [{'name':'mavg1', 'class':'MovingAverage', 'hours':1},{'name':'mavg2', 'class':'MovingAverage', 'hours':2}] - - result_address = "tcp://127.0.0.1:20202" - - mavg = MovingAverage(feed.feed_address, result_address, feed.sync_address, config['transforms'][0]) - mavg_proc = multiprocessing.Process(target=mavg.run) - logger.info("about to launch moving average") - mavg_proc.start() - - #merger = Merge(feed.feed_address, result_address, feed.sync_address, config) - #merger.run() - #merger_proc = multiprocessing.Process(target=merger.run) - #merger_proc.start() - - - #subscribe a client directly to the consolidated feed - #client = TestClient(feed.feed_address, feed.sync_address) - - logger.info("starting the client") - #subscribe a client to the transformed feed - client = TestClient(result_address, feed.sync_address, bind=True) - client.run() - - logger.info("feed has {pending} messages".format(pending=feed.data_buffer.pending_messages())) - assert(feed.data_buffer.pending_messages() == 0) - -def pubsub(): - proc1 = multiprocessing.Process(target=sub) - proc2 = multiprocessing.Process(target=pub) - proc1.start() - proc2.start() - -def sub(): - context = zmq.Context() - controller = context.socket(zmq.PULL) - controller.connect("tcp://127.0.0.1:10099") - #controller.setsockopt(zmq.SUBSCRIBE, '') - while True: - try: - message = controller.recv() - print message - except zmq.ZMQError as err: - if err.errno != zmq.EAGAIN: - raise err - -def pub(): - context = zmq.Context() - controller = context.socket(zmq.PUSH) - controller.bind("tcp://127.0.0.1:10099") - while True: - controller.send("HELLO3") - - - -if __name__ == "__main__": - tornado.options.parse_command_line() - datafeed() - \ No newline at end of file +if __name__ == '__main__': + suite = unittest.TestLoader().loadTestsFromTestCase(MessagingTestCase) + unittest.TextTestRunner(verbosity=2).run(suite) \ No newline at end of file diff --git a/test/__init__.py b/test/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/test/testmessaging.py b/test/testmessaging.py new file mode 100644 index 00000000..ad127390 --- /dev/null +++ b/test/testmessaging.py @@ -0,0 +1,66 @@ +import unittest +import zmq +import logging +import tornado +from data.sources.equity import * +from data.feed import * +from data.transforms import MergedTransformsFeed, MovingAverage + +from qbt_client import TestClient + + +class MessagingTestCase(unittest.TestCase): + + def setUp(self): + tornado.options.parse_command_line() + self.total_data_count = 800 + self.feed_config = {'emt1':{'sid':133, 'class':'RandomEquityTrades', 'count':400}, + 'emt2':{'sid':134, 'class':'RandomEquityTrades', 'count':400}} + self.feed = DataFeed(self.feed_config) + self.feed_proc = multiprocessing.Process(target=self.feed.run) + + self.config = {} + self.config['name'] = '**merged feed**' + self.config['transforms'] = [{'name':'mavg1', 'class':'MovingAverage', 'hours':1},{'name':'mavg2', 'class':'MovingAverage', 'hours':2}] + + def dtest_client(self): + #subscribe a client to the transformed feed + client = TestClient(self.feed, self.feed.feed_address) + + feed_proc = multiprocessing.Process(target=self.feed.run) + feed_proc.start() + + + client.run() + self.assertEqual(self.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages, found {n} remaining.".format(n=self.feed.data_buffer.pending_messages())) + self.assertEqual(self.total_data_count, client.received_count, "The client should have received ({n}) the same number of messages as the feed sent ({m}).".format(n=client.received_count, m=self.total_data_count)) + + + def dtest_moving_average_to_client(self): + mavg = MovingAverage(self.feed, self.config['transforms'][0]) + mavg_proc = multiprocessing.Process(target=mavg.run) + mavg_proc.start() + + client = TestClient(self.feed, mavg.result_address, bind=True) + + feed_proc = multiprocessing.Process(target=self.feed.run) + feed_proc.start() + + client.run() + self.assertEqual(self.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages.") + self.assertEqual(self.total_data_count, client.received_count, "The client should have received the same number of messages as the feed sent.") + + def test_merged_to_client(self): + merger = MergedTransformsFeed(self.feed, self.config) + merger_proc = multiprocessing.Process(target=merger.run) + merger_proc.start() + + client = TestClient(self.feed, merger.result_address) + + feed_proc = multiprocessing.Process(target=self.feed.run) + feed_proc.start() + + client.run() + self.assertEqual(self.feed.data_buffer.pending_messages(), 0, "The feed should be drained of all messages.") + self.assertEqual(self.total_data_count, client.received_count, "The client should have received the same number of messages as the feed sent.") + \ No newline at end of file