<commit_before>import sys
import logging
import urllib.parse
import asyncio
import bson
from bson.objectid import ObjectId
from motor.motor_asyncio import AsyncIOMotorClient


log = logging.getLogger(__name__)


@asyncio.coroutine
def make_connection(db):
    # TODO: Forking
    # log.info("???")
    clients = {}
    loop = asyncio.get_event_loop()
    cursor = db.items.find({})
    while (yield from cursor.fetch_next):
        item = cursor.next_object()
        host, port = item['address'], item['port']
        # task = asyncio.Task(check_tcp_port(host, port))
        task = asyncio.ensure_future(check_tcp_port(host, port))

        clients[task] = (host, port)

        def client_done(task):
            del clients[task]
            log.info("Client Task Finished")
            if len(clients) == 0:
                log.info("clients is empty, stopping loop.")
                loop = asyncio.get_event_loop()
                loop.stop()

        log.info("New Client Task")
        task.add_done_callback(client_done)


@asyncio.coroutine
def check_tcp_port(host, port):
    try:
        # Wait for 3 seconds, then raise TimeoutError
        conn = asyncio.open_connection(host, port)
        reader, writer = yield from asyncio.wait_for(
            conn, timeout=3)
        log.info(
            "Connection is alive {} {}".format(host, port))
        writer.close()
    except asyncio.TimeoutError:
        log.info("Timeout, skipping {} {}".format(host, port))
    except ConnectionRefusedError:
        log.info(
            "Connection refused, skipping {} {}"
            .format(host, port)
        )


def run():
    import argparse
    parser = argparse.ArgumentParser()

    parser.add_argument(
        "--db",
        help="Default: mongodb://127.0.0.1/monitort",
        type=str,
        default="mongodb://127.0.0.1/monitort"
    )
    parser.add_argument(
        "--forks",
        help="Default: 4",
        type=int,
        default=4
    )

    args = parser.parse_args()

    db = AsyncIOMotorClient(args.db).items

    loop = asyncio.get_event_loop()
    main = asyncio.ensure_future(make_connection(db))
    loop.run_forever()

    log.info("End")
    # loop = asyncio.get_event_loop()
    # task = asyncio.ensure_future(check_tcp_port(db))
    # loop.run_until_complete(task)
    # loop.close()

if __name__ == "__main__":
    log = logging.getLogger("")<commit_msg>forking (multiple tasks)<commit_after>import sys
import logging
import urllib.parse
import asyncio
import bson
from bson.objectid import ObjectId
from motor.motor_asyncio import AsyncIOMotorClient


log = logging.getLogger(__name__)


@asyncio.coroutine
def make_connection(db, forks=4):
    clients = {}
    loop = asyncio.get_event_loop()
    cursor = db.items.find({})
    for frk in range(forks):
        task = asyncio.ensure_future(check_tcp_port(cursor))

        clients[task] = frk

        def client_done(task):
            del clients[task]
            log.info("Client Task Finished")
            if len(clients) == 0:
                log.info("clients is empty, stopping loop.")
                loop = asyncio.get_event_loop()
                loop.stop()

        log.info("New Client Task")
        task.add_done_callback(client_done)


@asyncio.coroutine
def check_tcp_port(cursor):
    while (yield from cursor.fetch_next):
        item = cursor.next_object()
        host, port = item['address'], item['port']
        try:
            # Wait for 3 seconds, then raise TimeoutError
            conn = asyncio.open_connection(host, port)
            reader, writer = yield from asyncio.wait_for(
                conn, timeout=3)
            log.info(
                "Connection is alive {} {}".format(host, port))
            writer.close()
        except asyncio.TimeoutError:
            log.info("Timeout, skipping {} {}".format(host, port))
            continue
        except ConnectionRefusedError:
            log.info(
                "Connection refused, skipping {} {}"
                .format(host, port)
            )
            continue


def run():
    import argparse
    parser = argparse.ArgumentParser()

    parser.add_argument(
        "--db",
        help="Default: mongodb://127.0.0.1/monitort",
        type=str,
        default="mongodb://127.0.0.1/monitort"
    )
    parser.add_argument(
        "--forks",
        help="Default: 4",
        type=int,
        default=4
    )

    args = parser.parse_args()

    db = AsyncIOMotorClient(args.db).items

    loop = asyncio.get_event_loop()
    main = asyncio.ensure_future(make_connection(db, args.forks))
    loop.run_forever()


if __name__ == "__main__":
    log = logging.getLogger("")
{'fraction_non_alphanumeric': 0.1186513017498933, 'fraction_numerical': 0.007469056764831413, 'mean_word_length': 2.2845129642606867, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3666874', 'n_tokens_mistral': 1475, 'n_tokens_neox': 1414, 'n_words': 301}