Miksi tämä funktio ei päästä tausta-threadia loppuun asti?

[Jean_Sipulius]

Hiljaista on ohjelmointirintamalla, joten laitetaan taas uusi viesti. Laitoin aiemmin noista äänikaappauksista kysymyksiä ja tämä hipoo hiukan niitä, mutta kuitenkin eri asiaa.

Minulla on koodi, joka pyörii tausta-threadissa. Tämä threadi vastaa siitä, että serveri kuuntelee tulevaa ääntä jatkuvasti.

Käytännössä siis StartServer() funktio ajetaan tausta-triidissä ja se tuhoaa itse itsensä heti kun ajettava funktio on loppunut. Tietona sen verran, että serverin pystytys toimii ja voin JOINATA tähän toisella tietokoneella ja kun lähetän paketissa tiedon packet << endOfStream niin background lambda menee läpi ja thread suljetaan kunnollisesti.

Ongelma on siinä, en saa tuota pirun ikuista looppia
// Start receiving audio data
receiveLoop();

loppumaan millään tältä samalta koneelta! Haluaisin siis käynnistää ja sulkea serverin paikallisesti. Olen koittanut vaikka mitä. Jopa sen, että koko receiveLoop() on täysin TYHJÄ. Siltikään backgroundlambda ei pääse läpi. Mutta välittömästi jos siellä on koodia ja annan verkon yli tuon endOfStream komennon lambda pääsee läpi jälleen.

Koitin tehdä jopa niin, että loin tähän rinnalle SoundRecorderin ja heti kun se käynnistyy, se tuhoutuu funktion loputtua. Tuhotessaan tämä lähettää saman endOfStreamin serverille, mutta jostain syystä se ei auta.

UE_LOG(LogTemp, Warning, TEXT("Exit While Loop")); tuohon asti en pääse ollenkaan. Enkä tajua miksi. Tuossahan selkeästi on koodissa, että looppaa kunnes Canrun on totta while (Canrun)

Onko ideoita miksi ohjelma käyttäytyy näin? Miksi ohjelma ei lopu jos muutan tuota muuttujaa omalta koneelta ja miksi se loppuu jos lähetän ulkopuolelta tiedon? Tein jopa niin, että käynnistin erillisen puheohjelman samalta koneelta, josta yhdistin tähän serveriin. Silloinkin se toimii, joten mitään saman IP:n tarkistusongelmaa ei ole.

Osaisiko joku auttaa. Olisin tuhannesti kiitollinen. Tuo Voice chatti muuten jo toimii, mutta sitä ei tosiaan saa sammutettua . Kyselkää jos on epäselvää niin vastailen mahd. nopeasti.

// Fill out your copyright notice in the Description page of Project Settings.

#pragma once

#include "CoreMinimal.h"
#include "UObject/NoExportTypes.h"
#include "SFML/Audio.hpp"
#include <SFML/Network.hpp>
#include <list>
#include "HAL/ThreadSafeBool.h"
#include "Public/TimerManager.h"
#include "Engine/World.h"
#include "sfmlServer.generated.h"






////////////////////////////////////////////////////////////
/// Customized sound stream for acquiring audio data
/// from the network
////////////////////////////////////////////////////////////
class NetworkAudioStream : public sf::SoundStream
{
public:
    const sf::Uint8 audioData = 1;
    const sf::Uint8 endOfStream = 2;
    ////////////////////////////////////////////////////////////
    /// Default constructor
    ///
    ////////////////////////////////////////////////////////////
    NetworkAudioStream() :
        m_offset(0),
        m_hasFinished(false)
    {
        // Set the sound parameters
        initialize(1, 44100);
    }

    ////////////////////////////////////////////////////////////
    /// Run the server, stream audio data from the client
    ///
    ////////////////////////////////////////////////////////////
    void start(unsigned short port)
    {
        // Listen to the given port for incoming connections
        if (m_listener.listen(port) != sf::Socket::Done)
            return;
        UE_LOG(LogTemp, Warning, TEXT("Server Started at port %i"), port);

        selector.add(m_listener);

        // Start playback
        play();

        // Start receiving audio data
        receiveLoop();
    }

private:

    ////////////////////////////////////////////////////////////
    /// /see SoundStream::OnGetData
    ///
    ////////////////////////////////////////////////////////////
    virtual bool onGetData(sf::SoundStream::Chunk& data)
    {
        // We have reached the end of the buffer and all audio data have been played: we can stop playback
        if ((m_offset >= m_samples.size()) && m_hasFinished)
            return false;

        // No new data has arrived since last update: wait until we get some
        while ((m_offset >= m_samples.size()) && !m_hasFinished)
            sf::sleep(sf::milliseconds(10));

        // Copy samples into a local buffer to avoid synchronization problems
        // (don't forget that we run in two separate threads)
        {
            sf::Lock lock(m_mutex);
            m_tempBuffer.assign(m_samples.begin() + m_offset, m_samples.end());
        }

        // Fill audio data to pass to the stream
        data.samples = &m_tempBuffer[0];
        data.sampleCount = m_tempBuffer.size();

        // Update the playing offset
        m_offset += m_tempBuffer.size();
        UE_LOG(LogTemp, Warning, TEXT("Getting Data"));
        return true;
    }

    ////////////////////////////////////////////////////////////
    /// /see SoundStream::OnSeek
    ///
    ////////////////////////////////////////////////////////////
    virtual void onSeek(sf::Time timeOffset)
    {
        m_offset = timeOffset.asMilliseconds() * getSampleRate() * getChannelCount() / 1000;
    }

    ////////////////////////////////////////////////////////////
    /// Get audio data from the client until playback is stopped
    ///
    ////////////////////////////////////////////////////////////
    void receiveLoop()
    {  
        while (Canrun)
        {
            // Make the selector wait for data on any socket
            if (selector.wait())
            {
                // Test the listener
                if (selector.isReady(m_listener))
                {
                    // The listener is ready: there is a pending connection
                    sf::TcpSocket* client = new sf::TcpSocket;
                    if (m_listener.accept(*client) == sf::Socket::Done)
                    {  
                        // Add the new client to the clients list
                        clients.push_back(client);
                        // Add the new client to the selector so that we will
                        // be notified when he sends something
                        selector.add(*client);
                        UE_LOG(LogTemp, Warning, TEXT("New Client Joined to chat"));
                    }
                    else
                    {
                        // Error, we won't get a new connection, delete the socket
                        delete client;
                    }
                }
                else
                {
                    // The listener socket is not ready, test all other sockets (the clients)
                    for (std::list<sf::TcpSocket*>::iterator it = clients.begin(); it != clients.end(); ++it)
                    {
                        sf::TcpSocket& client = **it;
                        if (selector.isReady(client))
                        {
                            // The client has sent some data, we can receive it
                            sf::Packet packet;
                            if (client.receive(packet) == sf::Socket::Done)
                            {
                                // Get waiting audio data from the network
                                //sf::Packet packet;
                                //if (client.receive(packet) != sf::Socket::Done)
                                //   break;

                                // Extract the message ID
                                sf::Uint8 id;
                                packet >> id;


                                if (id == endOfStream)
                                {
                                    // End of stream reached: we stop receiving audio data
                                    //   std::cout << "Audio data has been 100% received!" << std::endl;
                                    UE_LOG(LogTemp, Warning, TEXT("Audio Stream successfully finished"));
                                    stop();
                                    Canrun = false;
                                }

                            else if (id == audioData)
                                {
                                    // Extract audio samples from the packet, and append it to our samples buffer
                                    const sf::Int16* samples = reinterpret_cast<const sf::Int16*>(static_cast<const char*>(packet.getData()) + 1);
                                    std::size_t      sampleCount = (packet.getDataSize() - 1) / sizeof(sf::Int16);

                                    // Don't forget that the other thread can access the sample array at any time
                                    // (so we protect any operation on it with the mutex)
                                    {
                                        sf::Lock lock(m_mutex);
                                        std::copy(samples, samples + sampleCount, std::back_inserter(m_samples));
                                    }


                                    // Send to all clients.
                                 
                                    for (std::list<sf::TcpSocket*>::iterator it = clients.begin(); it != clients.end(); ++it)
                                    {
                                        sf::TcpSocket& client2 = **it;
                                        // Don't send same voice to speaker.
                                        if (client2.getRemoteAddress() != client.getRemoteAddress())
                                        {
                                            client2.send(packet);
                                        }
                                    }
                                }
                         
 
                                else
                                {
                                    // Something's wrong...
                                    UE_LOG(LogTemp, Warning, TEXT("Invalid packet received..."));
                                    stop();
                                    Canrun = false;

                                }
                            }
                        }
                    }
                }
            }
        }
        UE_LOG(LogTemp, Warning, TEXT("Exit While Loop"));
    }

    ////////////////////////////////////////////////////////////
    // Member data
    ////////////////////////////////////////////////////////////
    sf::TcpListener        m_listener;
    sf::TcpSocket          m_client;
    // Create a list to store the future clients
    std::list<sf::TcpSocket*> clients;
    // Create a selector
    sf::SocketSelector selector;

    sf::Mutex              m_mutex;
    std::vector<sf::Int16> m_samples;
    std::vector<sf::Int16> m_tempBuffer;
    std::size_t            m_offset;
    bool                   m_hasFinished;
    bool Canrun = true;
};





// TÄÄ LUOTIIN JÄLKIKÄTEEN. KOITIN TÄSTÄ ANTAA TUON ENDOFSTREAM KOMENNON
// DISCONNECTER
////////////////////////////////////////////////////////////
/// Specialization of audio recorder for sending recorded audio
/// data through the network
////////////////////////////////////////////////////////////
class Disconnecter : public sf::SoundRecorder
{
public:

    ////////////////////////////////////////////////////////////
    /// Constructor
    ///
    /// \param host Remote host to which send the recording data
    /// \param port Port of the remote host
    ///
    ////////////////////////////////////////////////////////////
    Disconnecter(const sf::IpAddress& host, unsigned short port) :
        m_host(host),
        m_port(port)
    {
    }

    ////////////////////////////////////////////////////////////
    /// Destructor
    ///
    /// \see SoundRecorder::~SoundRecorder()
    ///
    ////////////////////////////////////////////////////////////
    ~Disconnecter()
    {
        // Make sure to stop the recording thread
        stop();
    }

private:

    ////////////////////////////////////////////////////////////
    /// \see SoundRecorder::onStart
    ///
    ////////////////////////////////////////////////////////////
    virtual bool onStart()
    {
        if (m_socket.connect(m_host, m_port) == sf::Socket::Done)
        {
            return true;
        }
        else
        {
            return false;
        }
    }





    ////////////////////////////////////////////////////////////
    /// \see SoundRecorder::onProcessSamples
    ///
    ////////////////////////////////////////////////////////////
    virtual bool onProcessSamples(const sf::Int16* samples, std::size_t sampleCount)
    {
        UE_LOG(LogTemp, Warning, TEXT("HMM"));

        // Pack the audio samples into a network packet
        sf::Packet packet;
        packet << audioData;
        packet.append(samples, sampleCount * sizeof(sf::Int16));

        // Send the audio packet to the server
        return m_socket.send(packet) == sf::Socket::Done;
    }

    ////////////////////////////////////////////////////////////
    /// \see SoundRecorder::onStop
    ///
    ////////////////////////////////////////////////////////////
    virtual void onStop()
    {
        UE_LOG(LogTemp, Warning, TEXT("Jaaha"));
        // Send a "end-of-stream" packet

        for (int i = 0; i < 10; i++)
        {
// Lähetä muutama äänifilu.
            sf::Packet packet;
            packet << audioData;
            m_socket.send(packet);
        }
// Lähetä sulje yhteys
        sf::Packet packet;
        packet << endOfStream;
        m_socket.send(packet);

        // Close the socket
        m_socket.disconnect();
    }

    ////////////////////////////////////////////////////////////
    // Member data
    ////////////////////////////////////////////////////////////
    sf::IpAddress  m_host;   ///< Address of the remote host
    unsigned short m_port;   ///< Remote port
    sf::TcpSocket  m_socket; ///< Socket used to communicate with the server
    const sf::Uint8 audioData = 1;
    const sf::Uint8 endOfStream = 2;
    const sf::Uint8 Fail = 3;
};




// Pääluokka

UCLASS()
class SFMLTEST_API UsfmlServer : public UObject
{
    GENERATED_BODY()

public:
    NetworkAudioStream audioStream;
 
    UFUNCTION(BlueprintCallable)
    void StartServer(int32 port);

    UFUNCTION(BlueprintCallable)
    void StopServer();  
};

.CPP

// Fill out your copyright notice in the Description page of Project Settings.

#include "sfmlServer.h"
#include "LambdaRunnable.h"

FLambdaRunnable* LambdaServer;


////////////////////////////////////////////////////////////
/// Launch a server and wait for incoming audio data from
/// a connected client
///
////////////////////////////////////////////////////////////
void UsfmlServer::StartServer(int32 port)
{
    LambdaServer = FLambdaRunnable::RunLambdaOnBackGroundThread([&]()
    {
        // Build an audio stream to play sound data as it is received through the network
        audioStream.start(port);

        // Loop until the sound playback is finished
        while (audioStream.getStatus() != sf::SoundStream::Stopped)
        {
            // Leave some CPU time for other threads
            sf::sleep(sf::milliseconds(100));
            //Initial wait before starting
            FPlatformProcess::Sleep(100);
        }
    });
}

void UsfmlServer::StopServer()
{
    Disconnecter Disconnect("localhost", 2435);
    Disconnect.start();
}

 

[OlviA]

Koodi varmaankin jumittaa tässä kohtaa, kunnes saa dataa

// Make the selector wait for data on any socket
if (selector.wait())

 

[Jean_Sipulius]

Koodi varmaankin jumittaa tässä kohtaa, kunnes saa dataa

// Make the selector wait for data on any socket
if (selector.wait())

Koitan tänään vielä säätää. Ideoita mitä koittaa? Mitän ideaa miksei ohjelma lopu jos tuo funktio on tyhjä? Käytän SFML kirjastoa.

 

[OlviA]

Anna sille parmetrina timeout.

 

[Jean_Sipulius]

Anna sille parmetrina timeout.

Koitin sitä jo aiemmin. Ei tuntunut auttavan asiaan. Hmm.

 

[Jean_Sipulius]

Löysin syyllisen, mutta en tiedä mikä tämän aiheuttaa. Syyllinen on se, että stop() funktiota ei ikinä ajeta, mikäli luon clientin ja yhdistän sen serveriin suoraan visual studion koodin kautta. Jos luon serverin ja avaan erillisen ohjelman, josta yhdistän ja suljen yhteyden serveri toimii. Testasin sitä näin:

 if (id == endOfStream)
  {
        // End of stream reached: we stop receiving audio data
        //   std::cout << "Audio data has been 100% received!" << std::endl;
        stop();
        Canrun = false;
        UE_LOG(LogTemp, Warning, TEXT("Audio Stream successfully finished"));
    }

Eli en ikinä saa logia, jos luon ja yhdistän paikallisesti. Mutta jos luon ja yhdistän eri ohjelman kautta, vaikkakin samasta IP:stä saan tuon login. Tämä myös selittää miksi ohjelma ei ikinä lopu vaikka koko koodipaikka on tyhjä. Ilman stop() komentoa, ohjelma on käytännössä ikuisesti päällä. Nyt en vaan tajua, miksi stop() komento ei toimi ilman ulkopuolista dataa verkon yli.

 

[OlviA]

Se sun koodi ei pääse siitä "if (selector.wait())" -riviltä eteenpäin, ellei ulkoapäin oteta yhteyttä.
Funktion wait() tulee palauttaa, joko true tai false, jotta tuosta voidaan jatkaa.

True palautuu ainostaan silloin, kun monitoroitu socket on vastaanottanut jotain dataa.
False vain siinä tapauksessa, että olet määrittänyt sille timeoutin ja tuo timeout on kulunut.
Kaikissa muissa tapauksissa koodi jumittaa tuolla rivillä maailman tappiin.

Kaikki mitä sen if-lauseen sisällä on, toteutetaan vain siinä tapauksessa, että funktio palauttaa true, eli socket on vastaanottanut dataa.
Joko nyt tajuat, miksi stop-komento ei toimi ilman ulkopuolista dataa?

Lisää se timeout ja else haara, joka käsittelee tapaukset, jolloin kukaan tulekaan kolkuttelemaan sockettejasi:

if (selector.wait(sf::seconds(10)))
{
    //vastaanota data
}
else
{
    //logita timeout reached jne...
}

Kannattaa muuten opetella debuggerin käyttö.

 

[Jean_Sipulius]

Se sun koodi ei pääse siitä "if (selector.wait())" -riviltä eteenpäin, ellei ulkoapäin oteta yhteyttä.
Funktion wait() tulee palauttaa, joko true tai false, jotta tuosta voidaan jatkaa.

True palautuu ainostaan silloin, kun monitoroitu socket on vastaanottanut jotain dataa.
False vain siinä tapauksessa, että olet määrittänyt sille timeoutin ja tuo timeout on kulunut.
Kaikissa muissa tapauksissa koodi jumittaa tuolla rivillä maailman tappiin.

Kaikki mitä sen if-lauseen sisällä on, toteutetaan vain siinä tapauksessa, että funktio palauttaa true, eli socket on vastaanottanut dataa.
Joko nyt tajuat, miksi stop-komento ei toimi ilman ulkopuolista dataa?

Lisää se timeout ja else haara, joka käsittelee tapaukset, jolloin kukaan tulekaan kolkuttelemaan sockettejasi:

if (selector.wait(sf::seconds(10)))
{
    //vastaanota data
}
else
{
    //logita timeout reached jne...
}

Kannattaa muuten opetella debuggerin käyttö.

Kiitos vastauksesta. Ongelma ei ole tässä tapauksessa tuo, että selectori jää odottamaan vaan jokin muu. Kyselin tästä jo tuolla SFML foorumilla ja odottelen vastausta. Laitan saman videon tännekin jonka tein. Siinä näkyy mitä ratkaisua olen koittanut ja miten ohjelma toimii näillä ratkaisuilla. Jos tuon SFML:n sielunelämä on tuttua niin ehdotuksia otetaan vastaan. Itse lyön veikkauksen, että tuolla jää joku SFML kirjaston oma tausta tread pyörimään ja sen takia se ei sulje ohjelmaa. Tausta-triidi sulkeutuu vain jos EndofStream viesti menee perille.

Tuossa, missä suljen serverin käytän juurikin semmosta hacky-tapaa, jossa luon uuden clientin joka yhdistää servuun ja lähettää EndofStream viestin, joka sulkee servun. Mutta tuo tapa on aika huono ja ei ollenkaan käytännöllinen. Video ongelmaan löytyy täältä:


edit: Homma ratkesi! Ongelma tosiaan oli threadi joka pyöri sf-kirjastossa! Se lähti päälle heti kun Play() funktiota kutsuttiin. Ohjelma kaipasi vain yhden lisärivin, joka oli se, että m_hasFinished on totta. Tämä sulki kirjaston threadin ja samoin sulkeutui Unrealissa rakennettu threadi.