def read(self, filename):
        nodes_queue = JoinableQueue(128)
        coords_queue = JoinableQueue(512)
        ways_queue = JoinableQueue(128)
        relations_queue = JoinableQueue(128)

        log_proc = self.logger()
        log_proc.start()

        marshal = True
        if self.merge:
            # merging needs access to unmarshaled data
            marshal = False

        estimates = {
            'coords': self.estimated_coords,
            'nodes': self.estimated_coords//50,
            'ways': self.estimated_coords//7,
            'relations': self.estimated_coords//1000,
        }

        coords_writer = CacheWriterProcess(coords_queue, self.cache.coords_cache,
            estimates['coords'], log=partial(log_proc.log, 'coords'),
            marshaled_data=marshal)
        coords_writer.start()

        nodes_writer = CacheWriterProcess(nodes_queue, self.cache.nodes_cache,
            estimates['nodes'], log=partial(log_proc.log, 'nodes'),
            marshaled_data=marshal)
        nodes_writer.start()


        ways_writer = CacheWriterProcess(ways_queue, self.cache.ways_cache,
            estimates['ways'], merge=self.merge, log=partial(log_proc.log, 'ways'),
            marshaled_data=marshal)
        ways_writer.start()

        relations_writer = CacheWriterProcess(relations_queue, self.cache.relations_cache,
            estimates['relations'], merge=self.merge, log=partial(log_proc.log, 'relations'),
            marshaled_data=marshal)
        relations_writer.start()

        log_proc.message('coords: %dk nodes: %dk ways: %dk relations: %dk (estimated)' % (
            estimates['coords']/1000, estimates['nodes']/1000, estimates['ways']/1000,
            estimates['relations']/1000)
        )

        # keep one CPU free for writer proc on hosts with 4 or more CPUs
        pool_size = self.pool_size if self.pool_size < 4 else self.pool_size - 1

        parser = OSMParser(pool_size, nodes_callback=nodes_queue.put, coords_callback=coords_queue.put,
            ways_callback=ways_queue.put, relations_callback=relations_queue.put, marshal_elem_data=marshal)

        parser.nodes_tag_filter = self.mapper.tag_filter_for_nodes()
        parser.ways_tag_filter = self.mapper.tag_filter_for_ways()
        parser.relations_tag_filter = self.mapper.tag_filter_for_relations()

        parser.parse(filename)

        coords_queue.put(None)
        nodes_queue.put(None)
        ways_queue.put(None)
        relations_queue.put(None)
        coords_writer.join()
        nodes_writer.join()
        ways_writer.join()
        relations_writer.join()
        log_proc.stop()
        log_proc.join()
dataset/ETHPy150Open omniscale/imposm/imposm/reader.py/ImposmReader.read/original