IjkPlayerシリーズのメッセージループメカニズム

PS: 最近読んだゲーテの詩の一節：あなたが何をできるか、または何を夢見るかにかかわらず、始めてみてください、大胆さは才能、エネルギー、そして魔法の同義語です。

前の 2 つの記事では、JNI の基本知識と IjkPlayer プレーヤーの作成プロセスについて紹介しました：

この記事の主な内容は以下の通りです：

AVMessage と MessageQueue
メッセージキューの初期化
メッセージループの開始
メッセージループスレッド
メッセージループ関数
小結

AVMessage と MessageQueue#

まず、AVMessageとMessageQueueの 2 つの構造体の定義を見てみましょう：

// ff_ffmsg_queue.h
typedef struct AVMessage {
    int what;
    int arg1;
    int arg2;
    void *obj;
    void (*free_l)(void *obj);
    struct AVMessage *next;
} AVMessage;

typedef struct MessageQueue {
    AVMessage *first_msg, *last_msg;
    int nb_messages;
    int abort_request;
    SDL_mutex *mutex;
    SDL_cond *cond;

    AVMessage *recycle_msg;
    int recycle_count;
    int alloc_count;
} MessageQueue;

AVMessageとMessageQueueの定義と実装は ff_ffmsg_queue.h にあり、関連する操作関数は主にmsg_xxxとmsg_queue_xxxです。以下のようになります：

// AVMessage
void msg_xxx(AVMessage *msg)
// MessageQueue
void msg_queue_xxx(MessageQueue *q)

MessageQueueの重要な関数は以下の通りです：

// MessageQueueの初期化
void msg_queue_init(MessageQueue *q)
// MessageQueueのリセット
void msg_queue_flush(MessageQueue *q)
// q->abort_requestを0に設定してメッセージループmsg_loopが実行できるようにする
void msg_queue_start(MessageQueue *q)
// msg_queue_put_xxx系列の関数はすべてmsg_queue_put_privateを呼び出す
int msg_queue_put_private(MessageQueue *q, AVMessage *msg)
// MessageQueueから最初のメッセージを取得する
int msg_queue_get(MessageQueue *q, AVMessage *msg, int block)

メッセージキューの初期化#

メッセージキューの初期化は IjkPlayer プレーヤーの作成プロセスで行われ、その重要な関数呼び出しは以下の通りです：

IjkMediaPlayer_native_setup->ijkmp_android_create->ijkmp_create

ここでは、ijkmp_create関数からメッセージキューの初期化を見ていきます。

メッセージキューはFFPlayerに定義されているmsg_queueメンバーに対応しており、IjkMediaPlayer構造体の作成プロセスでffp_create関数が呼び出され、ffplayerが初期化されます。以下のようになります：

// ijkplayer.c
IjkMediaPlayer *ijkmp_create(int (*msg_loop)(void*)){
    IjkMediaPlayer *mp = (IjkMediaPlayer *) mallocz(sizeof(IjkMediaPlayer));
    if (!mp)
        goto fail;
    // FFPlayerを作成しmp->ffplayerを初期化
    mp->ffplayer = ffp_create();
    // mp->msg_loop
    mp->msg_loop = msg_loop;
    // ...
}

次に、ffp_createの実装を見てみましょう：

// ff_ffplay.c
FFPlayer *ffp_create(){
    // ...
    // メッセージキューの初期化
    msg_queue_init(&ffp->msg_queue);
    ffp->af_mutex = SDL_CreateMutex();
    ffp->vf_mutex = SDL_CreateMutex();
    // 内部でmsg_queue_flushを呼び出す
    ffp_reset_internal(ffp);
    // ...
    return ffp;
}

ffp_create関数内でmsg_queue_initが呼び出され、メッセージキューmsg_queueが初期化されます。ここでmsg_loopのabort_requestが 1 に設定され、後でメッセージループスレッドを起動する際にabort_requestが 0 に設定されます。

ffp_reset_internal内で内部的にmsg_queue_flushが呼び出され、msg_loopがリセットされ、これでメッセージキューmsg_loopの初期化が完了しました。

次にmp->msg_loop = msg_loopでメッセージループ関数の代入が完了し、ijkmp_createで渡されたメッセージループ関数はmessage_loopです。この関数は後の節で紹介します。これでmessage_loopの代入が完了しました。

関数ポインタmsg_loopは確かに関数ですが、msg_loopはいつ呼び出されるのでしょうか？

メッセージループの開始#

メッセージループの開始はmsg_queue_start関数で、その呼び出しフローは再生準備が始まるとき、つまりアプリケーション層がprepareAsyncを呼び出して再生の準備を開始する際にトリガーされます。関数呼び出しフローは以下の通りです：

Mermaid Loading...

ここでijkmp_prepare_async_l関数の実装を見てみましょう：

// ijkplayer.c
static int ijkmp_prepare_async_l(IjkMediaPlayer *mp){
    // ...
    ijkmp_change_state_l(mp, MP_STATE_ASYNC_PREPARING);

    // メッセージループを開始
    msg_queue_start(&mp->ffplayer->msg_queue);

    // msg_loopで解放される
    ijkmp_inc_ref(mp);
    mp->msg_thread = SDL_CreateThreadEx(&mp->_msg_thread, ijkmp_msg_loop, mp, "ff_msg_loop");
    // ...
    return 0;
}

明らかにmsg_queue_startが呼び出され、メッセージループが開始されます。msg_queue_startの関数実装を見てみましょう：

// ff_ffmsg_queue.h
inline static void msg_queue_start(MessageQueue *q){
    SDL_LockMutex(q->mutex);
    // 重要
    q->abort_request = 0;
    AVMessage msg;
    msg_init_msg(&msg);
    msg.what = FFP_MSG_FLUSH;
    msg_queue_put_private(q, &msg);
    SDL_UnlockMutex(q->mutex);
}

ここでabort_requestが 0 に設定され、メッセージAVMessageの入隊と出隊が許可されます。この関数を呼び出さなければメッセージループを完了することはできません。したがって、メッセージキューmsg_queueを初期化した後、メッセージループを開始するためにmsg_queue_startを呼び出す必要があります。

次にメッセージループ取得関数msg_queue_getの実装を見てみましょう。メッセージの取得はこの関数を通じてアプリケーション層に通知されます。パラメータblockが 1 の場合はブロック、0 の場合は非ブロックを意味します。ここで渡されるのは 1 であり、つまりメッセージキューmsg_queueにメッセージがない場合は、メッセージが追加されるのを待ってから処理を続けます。以下のようになります：

 // ff_ffmsg_queue.h
inline static int msg_queue_get(MessageQueue *q, AVMessage *msg, int block){
    AVMessage *msg1;
    int ret;
    SDL_LockMutex(q->mutex);
    for (;;) {
        // 中止
        if (q->abort_request) {
            ret = -1;
            break;
        }
        // キューの先頭メッセージを取得
        msg1 = q->first_msg;
        if (msg1) {// キュー内のメッセージを処理
            q->first_msg = msg1->next;
            if (!q->first_msg)
                q->last_msg = NULL;
            q->nb_messages--;
            *msg = *msg1;
            msg1->obj = NULL;
#ifdef FFP_MERGE
            av_free(msg1);
#else
            msg1->next = q->recycle_msg;
            q->recycle_msg = msg1;
#endif
            ret = 1;
            break;
        } else if (!block) {// 直接退出
            ret = 0;
            break;
        } else {// ブロック待機
            SDL_CondWait(q->cond, q->mutex);
        }
    }
    SDL_UnlockMutex(q->mutex);
    return ret;
}

上記のコードでは、q->abort_requestが 0 である場合にのみメッセージの取得ループが開始されます。これがmsg_queue_startを使用してメッセージループを開始する理由です。

ここでのメッセージループの開始は、メッセージが正常に入隊および出隊できることを保証するだけですが、まだ実際にメッセージループ関数msg_loopが実行されているわけではありません。

メッセージループスレッド#

前述の問題を思い出してください。msg_loopはいつ呼び出されるのでしょうか。その答えは、msg_loopはメッセージループスレッド内で呼び出されるということです。前述のijkmp_prepare_async_lを続けて見ていくと、この関数内でmsg_queue_startが呼び出された後、メッセージループスレッドが作成されます。以下のようになります：

// ijkplayer.c
static int ijkmp_prepare_async_l(IjkMediaPlayer *mp){
    // ...
    ijkmp_change_state_l(mp, MP_STATE_ASYNC_PREPARING);
    // メッセージループを開始
    msg_queue_start(&mp->ffplayer->msg_queue);
    // msg_loopで解放される
    ijkmp_inc_ref(mp);
    // メッセージループスレッドを作成
    mp->msg_thread = SDL_CreateThreadEx(&mp->_msg_thread, ijkmp_msg_loop, mp, "ff_msg_loop");
    // ...
    return 0;
}

上記のコードでは、SDL_CreateThreadExがスレッド名ff_msg_loopのスレッドを作成し、スレッド本体はijkmp_msg_loop関数を実行します。同時に、IjkMediaPlayer構造体のメンバーmsg_threadが代入されます。スレッドが作成されると、スレッド本体関数ijkmp_msg_loopが実行されます。以下のようになります：

// ijkplayer.c
static int ijkmp_msg_loop(void *arg){
    IjkMediaPlayer *mp = arg;
    // メッセージループ関数を呼び出す
    int ret = mp->msg_loop(arg);
    return ret;
}

ここでメッセージループ関数msg_loopの呼び出しが完了し、IjkPlayer のメッセージループが実際に開始されます。次に、メッセージがアプリケーション層にどのように送信されるかを見ていきます。

メッセージループ関数#

メッセージループ関数はプレーヤー作成時に、IjkMediaPlayer_native_setup内で渡されます。以下のようになります：

// ijkplayer_jni.c
static void IjkMediaPlayer_native_setup(JNIEnv *env, jobject thiz, jobject weak_this){
    MPTRACE("%s\n", __func__);
    // C層に対応するIjkMediaPlayerを作成
    IjkMediaPlayer *mp = ijkmp_android_create(message_loop);
    // ...
}

message_loopは最終的にIjkMediaPlayer構造体のメンバーmsg_loopに代入されます。前述のようにメッセージループスレッドmsg_thread内でメッセージループ関数msg_loopが呼び出されます。ここでのmessage_loopの実装は以下の通りです：

// ijkplayer_jni.c
static int message_loop(void *arg){
    // ...
    IjkMediaPlayer *mp = (IjkMediaPlayer*) arg;
    // 重要な関数message_loop_n
    message_loop_n(env, mp);
    // ...
}

次に、重要な関数message_loop_nの実装を見てみましょう：

// ijkplayer_jni.c
static void message_loop_n(JNIEnv *env, IjkMediaPlayer *mp){
    jobject weak_thiz = (jobject) ijkmp_get_weak_thiz(mp);
    JNI_CHECK_GOTO(weak_thiz, env, NULL, "mpjni: message_loop_n: null weak_thiz", LABEL_RETURN);
    while (1) {
        AVMessage msg;
        // MessageQueueからAVMessageを取得
        int retval = ijkmp_get_msg(mp, &msg, 1);
        if (retval < 0)
            break;

        // ブロック取得は0を返すべきではない
        assert(retval > 0);
        // 様々な再生イベントを処理
        switch (msg.what) {
        case FFP_MSG_PREPARED:
            MPTRACE("FFP_MSG_PREPARED:\n");
            // 重要な関数
            post_event(env, weak_thiz, MEDIA_PREPARED, 0, 0);
            break;
        // ...
        default:
            ALOGE("unknown FFP_MSG_xxx(%d)\n", msg.what);
            break;
        }
        // メモリリソースの回収
        msg_free_res(&msg);
    }
LABEL_RETURN:
    ;
}

メッセージループ関数内では、無限ループの中でijkmp_get_msgを通じてメッセージを取得し、post_eventを通じてアプリケーション層にメッセージを送信します：

// ijkplayer_jni.c
inline static void post_event(JNIEnv *env, jobject weak_this, int what, int arg1, int arg2){
    // postEventFromNativeを呼び出す
    J4AC_IjkMediaPlayer__postEventFromNative(env, weak_this, what, arg1, arg2, NULL);
}

関数post_eventは Java 層のpostEventFromNativeメソッドを呼び出し、メッセージを返します。以下のようになります：

@CalledByNative
private static void postEventFromNative(Object weakThiz, int what,
        int arg1, int arg2, Object obj) {
    if (weakThiz == null)
        return;

    @SuppressWarnings("rawtypes")
    IjkMediaPlayer mp = (IjkMediaPlayer) ((WeakReference) weakThiz).get();
    if (mp == null) {
        return;
    }

    if (what == MEDIA_INFO && arg1 == MEDIA_INFO_STARTED_AS_NEXT) {
        // 必要に応じてウェイクロックを取得し、クライアント側の状態を設定
        mp.start();
    }
    if (mp.mEventHandler != null) {
        Message m = mp.mEventHandler.obtainMessage(what, arg1, arg2, obj);
        // メッセージを送信
        mp.mEventHandler.sendMessage(m);
    }
}

postEventFromNativeは IjkPlayer から送信されたメッセージを受け取り、Messageに変換してEventHandlerに処理を委ねます。以下のようになります：

private static class EventHandler extends Handler {
    private final WeakReference<IjkMediaPlayer> mWeakPlayer;
    public EventHandler(IjkMediaPlayer mp, Looper looper) {
        super(looper);
        mWeakPlayer = new WeakReference<IjkMediaPlayer>(mp);
    }
    @Override
    public void handleMessage(Message msg) {
        IjkMediaPlayer player = mWeakPlayer.get();
        if (player == null || player.mNativeMediaPlayer == 0) {
            DebugLog.w(TAG,
                    "IjkMediaPlayer went away with unhandled events");
            return;
        }
        switch (msg.what) {
        case MEDIA_PREPARED:
            player.notifyOnPrepared();
            return;
        // ...
        default:
            DebugLog.e(TAG, "Unknown message type " + msg.what);
        }
    }
}

異なるメッセージタイプに基づいて処理され、上記はMEDIA_PREPAREDイベントで、最終的に対応するコールバックインターフェースに通知されます。ここでのmOnPreparedListenerは以下のようになります：

protected final void notifyOnPrepared() {
    if (mOnPreparedListener != null)
        // 再生準備完了イベント
        mOnPreparedListener.onPrepared(this);
}

これでメッセージループ関数の実行が完了しました。

小結#

この記事はネイティブ層から始まり、Java 層がメッセージを受け取るまでの流れを説明しました。IjkPlayer のメッセージループで最も重要なのはメッセージキューmsg_queueであり、プレーヤーの開始から終了までに発生する関連イベントメッセージはすべてこのキューに追加されます。メッセージループスレッドはメッセージを取り出し、通知を行います。メッセージがなければ、追加されるのを待ってメッセージループのプロセスを続行します。