**Apache Airflow version**:
2.0.1

**Kubernetes version (if you are using kubernetes)** (use `kubectl version`):
Not applicable

**Environment**:

- **Cloud provider or hardware configuration**: None
- **OS** (e.g. from /etc/os-release): Ubuntu
- **Kernel** (e.g. `uname -a`): 
- **Install tools**:
- **Others**:

**What happened**:

1. Ran airflow db init on mysql in parallel in 2 command lines. Only one command did the migrations, the other one waited. But connections were inserted twice. I would like them not to be added twice.
2. Ran airflow db init on postgres in parallel in 2 command lines. Both command lines started doing migrations on the same db in parallel. I would like one command to run, the other to wait.

**What you expected to happen**:

1. For MySQL. Connections and other config objects to be inserted only once.
2. For Postgres. Only one migration can be performed in the same time for the same db.

**How to reproduce it**:
Scenario 1:
Setup Airflow so that it uses MySQL.
Run `airflow init db` in 2 command lines, side by side.

Scenario 2:
Setup Airflow so that it uses Postgres.
Run `airflow init db` in 2 command lines, side by side.

**Anything else we need to know**:
This problem occurs every time.
airflow/utils/session.py
"""Find session index in function call parameter."""
    func_params = signature(func).parameters
    try:
        
        session_args_idx = tuple(func_params).index("session")
    except ValueError:
        raise ValueError(f"Function {func.__qualname__} has no `session` argument") from None
    return session_args_idx
def provide_session(func: Callable[..., RT]) -> Callable[..., RT]:
    """
    Function decorator that provides a session if it isn't provided.
    If you want to reuse a session or run the function as part of a
    database transaction, you pass it to the function, if not this wrapper
    will create one and close it for you.
    """
    session_args_idx = find_session_idx(func)
    @wraps(func)
    def wrapper(*args, **kwargs) -> RT:
        if "session" in kwargs or session_args_idx < len(args):
            return func(*args, **kwargs)
        else:
            with create_session() as session:
                return func(*args, session=session, **kwargs)
    return wrapper
@provide_session
@contextlib.contextmanager
def create_global_lock(session=None, pg_lock_id=1, lock_name='init', mysql_lock_timeout=1800):